text:
#
# Python script to run shadow3. Created automatically with ShadowTools.make_python_script_from_list().
#
import Shadow
import numpy

def run_shadow(grazing=22.3,shift=0.0,ymax=-0.2):

    # oe1.SIMAG = 4.29
    # oe1.SSOUR = 1.58 + shift
    # oe1.T_INCIDENCE =  90.0 - grazing #67.7
    # oe1.T_REFLECTION = 90.0 - grazing #67.7
    # oe1.THETA =        90.0 - grazing #67.7

    # y = beam.rays[:, 1]
    # ibad = numpy.where(y > ymax)
    # beam.rays[ibad, 6:9] = 0.0
    # beam.rays[ibad, 15:18] = 0.0

    #
    # Python script to run shadow3. Created automatically with ShadowTools.make_python_script_from_list().
    #
    import Shadow
    import numpy

    # write (1) or not (0) SHADOW files start.xx end.xx star.xx
    iwrite = 0

    #
    # initialize shadow3 source (oe0) and beam
    #
    beam = Shadow.Beam()
    oe0 = Shadow.Source()
    oe1 = Shadow.OE()
    oe2 = Shadow.OE()

    #
    # Define variables. See meaning of variables in:
    #  https://raw.githubusercontent.com/srio/shadow3/master/docs/source.nml
    #  https://raw.githubusercontent.com/srio/shadow3/master/docs/oe.nml
    #

    oe0.BENER = 1.9
    oe0.CONV_FACT = 1.0
    oe0.EPSI_X = 1.989e-09
    oe0.EPSI_Z = 3.007e-11
    oe0.FDISTR = 0
    oe0.FILE_TRAJ = b'/Users/srio/Oasys/xshwig.sha'
    oe0.FSOUR = 0
    oe0.FSOURCE_DEPTH = 0
    oe0.F_COLOR = 0
    oe0.F_PHOT = 0
    oe0.F_WIGGLER = 1
    oe0.HDIV1 = 1.0
    oe0.HDIV2 = 1.0
    oe0.IDO_VX = 0
    oe0.IDO_VZ = 0
    oe0.IDO_X_S = 0
    oe0.IDO_Y_S = 0
    oe0.IDO_Z_S = 0
    oe0.ISTAR1 = 5676561
    oe0.NCOL = 0
    oe0.NPOINT = 50000
    oe0.N_COLOR = 0
    oe0.PH1 = 1000.0
    oe0.PH2 = 1000.1
    oe0.POL_DEG = 0.0
    oe0.SIGMAX = 3.9e-05
    oe0.SIGMAY = 0.0
    oe0.SIGMAZ = 3.1e-05
    oe0.VDIV1 = 1.0
    oe0.VDIV2 = 1.0
    oe0.WXSOU = 0.0
    oe0.WYSOU = 0.0
    oe0.WZSOU = 0.0

    oe1.ALPHA = 90.0
    oe1.DUMMY = 100.0
    oe1.FCYL = 1
    oe1.FHIT_C = 1
    oe1.FMIRR = 1
    oe1.FWRITE = 1
    oe1.F_DEFAULT = 0
    oe1.RLEN1 = 0.3
    oe1.RLEN2 = 0.4
    oe1.RWIDX1 = 0.65
    oe1.RWIDX2 = 0.65
    oe1.SIMAG = 4.29
    oe1.SSOUR = 1.58
    oe1.THETA = 67.7
    oe1.T_IMAGE = 0.0
    oe1.T_INCIDENCE = 67.7
    oe1.T_REFLECTION = 67.7
    oe1.T_SOURCE = 1.58

    oe1.SIMAG = 4.29
    oe1.SSOUR = 1.58 + shift
    oe1.T_INCIDENCE =  90.0 - grazing #67.7
    oe1.T_REFLECTION = 90.0 - grazing #67.7
    oe1.THETA =        90.0 - grazing #67.7

    oe2.ALPHA = 90.0
    oe2.DUMMY = 100.0
    oe2.FCYL = 1
    oe2.FMIRR = 1
    oe2.FWRITE = 1
    oe2.F_DEFAULT = 0
    oe2.SIMAG = 3.11
    oe2.SSOUR = 2.76
    oe2.THETA = 45.0
    oe2.T_IMAGE = 3.11
    oe2.T_INCIDENCE = 45.0
    oe2.T_REFLECTION = 45.0
    oe2.T_SOURCE = 1.18

    # Run SHADOW to create the source

    if iwrite:
        oe0.write("start.00")

    beam.genSource(oe0)

    y = beam.rays[:, 1]
    ibad = numpy.where(y < ymax)
    beam.rays[ibad, 6:9] = 0.0
    beam.rays[ibad, 15:18] = 0.0

    if iwrite:
        oe0.write("end.00")
        beam.write("begin.dat")

    #
    # run optical element 1
    #
    print("    Running optical element: %d" % (1))
    if iwrite:
        oe1.write("start.01")

    beam.traceOE(oe1, 1)

    if iwrite:
        oe1.write("end.01")
        beam.write("star.01")

    #
    # run optical element 2
    #
    print("    Running optical element: %d" % (2))
    if iwrite:
        oe2.write("start.02")

    beam.traceOE(oe2, 2)

    if iwrite:
        oe2.write("end.02")
        beam.write("star.02")

    # Shadow.ShadowTools.plotxy(beam, 1, 3, nbins=101, nolost=1, title="Real space")
    # Shadow.ShadowTools.plotxy(beam,1,4,nbins=101,nolost=1,title="Phase space X")
    # Shadow.ShadowTools.plotxy(beam,3,6,nbins=101,nolost=1,title="Phase space Z")

    return beam, oe1

if __name__ == "__main__":

    from srxraylib.plot.gol import plot, set_qt
    from srxraylib.util.h5_simple_writer import H5SimpleWriter

    set_qt()

    Grazing = numpy.linspace(12,16,80)
    # Grazing = numpy.linspace(15, 88, 50)
    Radius = numpy.zeros_like(Grazing)
    Fwhm = numpy.zeros_like(Grazing)
    Std = numpy.zeros_like(Grazing)
    Peak = numpy.zeros_like(Grazing)

    h = H5SimpleWriter.initialize_file("ir_scan_wiggler.h5")

    for i,grazing in enumerate(Grazing):

        beam,oe1 = run_shadow(grazing=grazing,shift=-0.4,ymax=0.2)
        # Shadow.ShadowTools.plotxy(beam, 1, 3, nbins=101, nolost=1, title="Real space")

        tkt = beam.histo1(1,nbins=201,nolost=1) # xrange=[-4000e-6,4000e-6],
        Radius[i] = oe1.RMIRR
        try:
            print(grazing,oe1.RMIRR,1e6*tkt["fwhm"])
            Fwhm[i] = 1e6*tkt["fwhm"]
        except:
            Fwhm[i] = -1
        Std[i] = 1e6 * beam.get_standard_deviation(1,nolost=1)
        Peak[i] = tkt["histogram"].max() / ((tkt["bin_center"])[1] - (tkt["bin_center"])[0] )

        h.create_entry("iteration %f" % grazing, nx_default="histogram")
        h.add_dataset(1e6*tkt["bin_center"],tkt["histogram"], dataset_name="histogram",entry_name="iteration %f" % grazing,
                      title_x="X / um",title_y="intensity / a.u.")

    h.create_entry("scan results", nx_default="Fwhm")
    h.add_dataset(Grazing, Fwhm, dataset_name="Fwhm", entry_name="scan results",
                  title_x="grazing angle / deg", title_y="fwhm / um")
    h.add_dataset(Grazing, Std, dataset_name="Std", entry_name="scan results",
                  title_x="grazing angle / deg", title_y="fwhm / um")
    h.add_dataset(Radius, Fwhm, dataset_name="Radius", entry_name="scan results",
                  title_x="radius / m", title_y="fwhm / um")
    h.add_dataset(Grazing, Peak, dataset_name="Peak", entry_name="scan results",
                  title_x="grazing angle / deg", title_y="peak")

    for key in tkt.keys():
        print(key)
    # Shadow.ShadowTools.plotxy(beam,1,4,nbins=101,nolost=1,title="Phase space X")
    # Shadow.ShadowTools.plotxy(beam,3,6,nbins=101,nolost=1,title="Phase space Z")

    plot(Radius,Fwhm,xtitle="R/m",ytitle="FWHM/um",show=False)
    plot(Grazing, Fwhm,Grazing, Std,xtitle="Graz/deg",ytitle="FWHM/um",legend=["fwhm","std"],show=False)
    plot(Grazing, Peak, xtitle="Graz/deg", ytitle="peak")


text:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on 23/10/17

@author: Maurizio Ferrari Dacrema
"""

from Utils.Base.Recommender_utils import check_matrix
from Utils.Base.BaseSimilarityMatrixRecommender import BaseUserSimilarityMatrixRecommender

from Utils.Base.IR_feature_weighting import okapi_BM_25, TF_IDF
import numpy as np

from Utils.Base.Similarity.Compute_Similarity import Compute_Similarity

class UserKNNCFRecommender(BaseUserSimilarityMatrixRecommender):
    """ UserKNN recommender"""

    RECOMMENDER_NAME = "UserKNNCFRecommender"

    FEATURE_WEIGHTING_VALUES = ["BM25", "TF-IDF", "none"]

    def __init__(self, URM_train, verbose = True):
        super(UserKNNCFRecommender, self).__init__(URM_train, verbose = verbose)

    def fit(self, topK=50, shrink=100, similarity='cosine', normalize=True, feature_weighting = "none", **similarity_args):

        self.topK = topK
        self.shrink = shrink

        if feature_weighting not in self.FEATURE_WEIGHTING_VALUES:
            raise ValueError("Value for 'feature_weighting' not recognized. Acceptable values are {}, provided was '{}'".format(self.FEATURE_WEIGHTING_VALUES, feature_weighting))

        if feature_weighting == "BM25":
            self.URM_train = self.URM_train.astype(np.float32)
            self.URM_train = okapi_BM_25(self.URM_train.T).T
            self.URM_train = check_matrix(self.URM_train, 'csr')

        elif feature_weighting == "TF-IDF":
            self.URM_train = self.URM_train.astype(np.float32)
            self.URM_train = TF_IDF(self.URM_train.T).T
            self.URM_train = check_matrix(self.URM_train, 'csr')

        similarity = Compute_Similarity(self.URM_train.T, shrink=shrink, topK=topK, normalize=normalize, similarity = similarity, **similarity_args)

        self.W_sparse = similarity.compute_similarity()
        self.W_sparse = check_matrix(self.W_sparse, format='csr')

ext: py

reason: import numpy


text:
import numpy as np

class Tensor:
    """Represents n-dimensional array (tensor)."""

    def __init__(self, ndarray: np.ndarray) -> None:
        """Ctor.

        Args:
            ndarray (np.ndarray): Elements of the tensor.
        """
        self.ndarray = ndarray
        self.grad = None

max_stars_repo_name: r-papso/pynet

max_issues_repo_name: r-papso/pynet

max_forks_repo_name: r-papso/pynet


text:
from numpy import std

def standard(data, ddof=0):
    v = std(data, ddof=ddof)
    return round(v, 3)

reason: from numpy


text:
import sys
import numpy as np

from OpenGL.GL import *
from OpenGL.GL import shaders
from OpenGL.GLUT import *

vao = None
vbo = None
shaderProgram = None

def readShaderFile(filename):
	with open('shader/' + filename, 'r') as myfile:
		return myfile.read()

def init():
	global shaderProgram
	global vao
	global vbo

	glClearColor(0, 0, 0, 0);

	vertex_code = readShaderFile('hello.vp')
	fragment_code = readShaderFile('hello.fp')

	# compile shaders and program
	vertexShader = shaders.compileShader(vertex_code, GL_VERTEX_SHADER)
	fragmentShader = shaders.compileShader(fragment_code, GL_FRAGMENT_SHADER)
	shaderProgram = shaders.compileProgram(vertexShader, fragmentShader)

	# Create and bind the Vertex Array Object
	vao = GLuint(0)
	glGenVertexArrays(1, vao)
	glBindVertexArray(vao)

	# Create and bind the Vertex Buffer Object
	vertices = np.array([[0, 0, 0, 0, 0, 1], [-1, -1, 0, 1, 1, 0], [1, -1, 0, 0, 1, 0]], dtype='f')
	vbo = glGenBuffers(1)
	glBindBuffer(GL_ARRAY_BUFFER, vbo)
	glBufferData(GL_ARRAY_BUFFER, vertices, GL_STATIC_DRAW)
	glVertexAttribPointer(0, 3, GL_FLOAT, False, 6 * sizeof(GLfloat), ctypes.c_void_p(0))  # first 0 is the location in shader
	glVertexAttribPointer(1, 3, GL_FLOAT, False, 6 * sizeof(GLfloat), ctypes.c_void_p(3*sizeof(GLfloat)))  # first 0 is the location in shader

	glEnableVertexAttribArray(0);  # 0=location do atributo, tem que ativar todos os atributos inicialmente sao desabilitados por padrao
	glEnableVertexAttribArray(1);  # 1=location do atributo, tem que ativar todos os atributos inicialmente sao desabilitados por padrao

	# Note that this is allowed, the call to glVertexAttribPointer registered VBO
	# as the currently bound vertex buffer object so afterwards we can safely unbind
	glBindBuffer(GL_ARRAY_BUFFER, 0)
	# Unbind VAO (it's always a good thing to unbind any buffer/array to prevent strange bugs)
	glBindVertexArray(0)

def display():
	global shaderProgram
	global vao

	glClear(GL_COLOR_BUFFER_BIT | GL_DEPTH_BUFFER_BIT)

	# load everthing back
	glUseProgram(shaderProgram)
	glBindVertexArray(vao)
	glBindBuffer(GL_ARRAY_BUFFER, vbo)

	# glDrawArrays( mode , first, count)
	glDrawArrays(GL_TRIANGLES, 0, 3)

	#clean things up
	glBindBuffer(GL_ARRAY_BUFFER, 0)
	glBindVertexArray(0)
	glUseProgram(0)

	glutSwapBuffers()  # necessario para windows!

def reshape(width, height):
	glViewport(0, 0, width, height)

if __name__ == '__main__':
	glutInit(sys.argv)
	glutInitContextVersion(3, 0)
	glutInitContextProfile(GLUT_CORE_PROFILE);
	glutInitDisplayMode(GLUT_DOUBLE | GLUT_RGB | GLUT_DEPTH)

	glutInitWindowSize(640, 640);
	glutCreateWindow(b'Hello world!')

	glutReshapeFunc(reshape)
	glutDisplayFunc(display)

	init()

	glutMainLoop()


text:
# USAGE
# python cluster_histograms.py --dataset dataset

# import the necessary packages
from pyimagesearch.descriptors.histogram import Histogram
from sklearn.cluster import KMeans
from imutils import paths
import numpy as np
import argparse
import cv2
import imutils

# construct the argument parse and parse the arguments
ap = argparse.ArgumentParser()
ap.add_argument("-d", "--dataset", required=True,
	help="path to the input dataset directory")
ap.add_argument("-k", "--clusters", type=int, default=2,
	help="# of clusters to generate")
args = vars(ap.parse_args())

# initialize the image descriptor along with the image matrix
desc = Histogram([8, 8, 8], cv2.COLOR_BGR2LAB)
data = []

# grab the image paths from the dataset directory
imagePaths = list(paths.list_images(args["dataset"]))
imagePaths = np.array(sorted(imagePaths))

# loop over the input dataset of images
for imagePath in imagePaths:
	# load the image, describe the image, then update the list of data
	image = cv2.imread(imagePath)
	hist = desc.describe(image)
	data.append(hist)

# cluster the color histograms
clt = KMeans(n_clusters=args["clusters"], random_state=42)
labels = clt.fit_predict(data)

# loop over the unique labels
for label in np.unique(labels):
	# grab all image paths that are assigned to the current label
	labelPaths = imagePaths[np.where(labels == label)]

	# placeholder for horizontal stack
	stack = []

	# loop over the image paths that belong to the current label
	for (i, path) in enumerate(labelPaths):
		# load the image, force size, and display it
		image = cv2.resize(cv2.imread(path), (200, 200))
		stack.append(image)

	# display the cluster
	cv2.imshow("Cluster #{}".format(label + 1), np.hstack(stack))

# wait for a keypress and then close all open windows
cv2.waitKey(0)
cv2.destroyAllWindows()

max_stars_repo_name: OddballSports-tv/obies-eyes

max_issues_repo_name: OddballSports-tv/obies-eyes

max_forks_repo_name: OddballSports-tv/obies-eyes


text:
# Based on the source code of GPyOpt/methods/modular_bayesian_optimization.py,
# which is licensed under the BSD 3-clause license

import time
import numpy as np
from GPyOpt.core.bo import BO
from GPyOpt.util.general import best_value
from GPyOpt.methods.modular_bayesian_optimization import ModularBayesianOptimization
from GPyOpt.plotting.plots_bo import plot_acquisition, plot_convergence

__all__ = ['EchoStateBO']

class EchoStateBO(BO):
    """
    Modular Bayesian optimization, edited for stepwise application in Echo State Networks.
    This class wraps the optimization loop around the different handlers.

    Adapted from ModularBayesianOptimization (Copyright (c) 2016, the GPyOpt Authors, BSD 3 license)

    :param model: GPyOpt model class.
    :param space: GPyOpt space class.
    :param objective: GPyOpt objective class.
    :param acquisition: GPyOpt acquisition class.
    :param evaluator: GPyOpt evaluator class.
    :param X_init: 2d numpy array containing the initial inputs (one per row) of the model.
    :param Y_init: 2d numpy array containing the initial outputs (one per row) of the model.
    :param cost: GPyOpt cost class (default, none).
    :param normalize_Y: whether to normalize the outputs before performing any optimization (default, True).
    :param model_update_interval: interval of collected observations after which the model is updated (default, 1).
    """

    def __init__(self, model, space, objective, acquisition, evaluator, X_init, Y_init=None, model_update_interval=1):

        self.initial_iter = True
        self.modular_optimization = True

        # Create optimization space
        super(EchoStateBO, self).__init__(
            model=model,
            space=space,
            objective=objective,
            acquisition=acquisition,
            evaluator=evaluator,
            X_init=X_init,
            Y_init=Y_init,
            cost=None,
            normalize_Y=False,  # Normalization done by RobustGPModel
            model_update_interval=model_update_interval)

    def run_target_optimization(self, target_score=0., max_iter=2048, max_time=np.inf, eps=1e-8, verbosity=True):
        """
        Runs Bayesian Optimization for a number 'max_iter' of iterations (after the initial exploration data)
        :param max_iter: exploration horizon, or number of acquisitions.
            If nothing is provided optimizes the current acquisition.
        :param max_time: maximum exploration horizon in seconds.
        :param eps: minimum distance between two consecutive x's to keep running the model.
        :param verbosity: flag to print the optimization results after each iteration (default, True).
        :param report_file: filename of the file where the results of the optimization are saved (default, None).
        """
        # Save the options to print and save the results
        self.verbosity = verbosity

        # Avoid stacking: create full X and Y
        X_full = np.full((max_iter, self.X.shape[1]), fill_value=np.nan, dtype=np.float32)
        Y_full = np.full((max_iter, 1), fill_value=np.inf, dtype=np.float32)

        # Setting up stop conditions
        self.eps = eps
        self.max_iter = max_iter
        self.max_time = max_time
        self.target = target_score

        # Initialize iterations and running time
        self.time_zero = time.time()
        self.cum_time = 0
        self.num_acquisitions = 0

        # Initial function evaluation and model fitting
        if self.X is not None and self.Y is None:
            samples = self.X.shape[0]
            Y, _ = self.objective.evaluate(self.X)
            X_full[:samples, :] = self.X  # Confusingly, self.X and self.Y here are initial samples
            Y_full[:samples, :] = Y
            self.num_acquisitions += samples

        # Overwrite X and Y with full arrays to prevent stacking (slow)
        self.X = X_full
        self.Y = Y_full

        first_time = True

        # Initialize time cost of the evaluations
        while (self.max_time > self.cum_time):

            # Update GP model
            self._update_model()

            # Inform user
            if first_time:
                if self.verbosity:
                    print("Model initialization done.", '\n')
                    print(self.model.model, '\n')
                    print(self.model.model.kern.lengthscale, '\n')
                first_time = False

            # Update and optimize acquisition and compute the exploration level in the next evaluation
            self.suggested_sample = self._compute_next_evaluations()

            # Augment X
            i = self.num_acquisitions
            sample_size = self.suggested_sample.shape[0]
            space_left = self.X.shape[0] - i
            allocation = np.clip(sample_size, 0, space_left)
            self.X[i:i + allocation, :] = self.suggested_sample[:allocation]

            # Evaluate *f* in X, augment Y
            Y_new = self.evaluate_objective()
            self.Y[i:i + allocation, :] = Y_new[:allocation]

            # Update current evaluation time and function evaluations
            self.cum_time = time.time() - self.time_zero
            distance = self._distance_last_evaluations(sample_size)
            self.num_acquisitions += allocation

            # Target check
            if np.any(Y_new <= self.target):
                if self.verbosity:
                    pass
                print('Target reached at iteration', self.num_acquisitions)
                break

            # Check maximum iterations
            if self.num_acquisitions >= self.max_iter:
                if self.max_iter > 0 and self.verbosity:
                    print('Maximum iterations reached')
                break

            # Convergence
            if distance <= self.eps:
                if self.verbosity:
                    pass
                print('Converged at iteration', self.num_acquisitions)
                break

        # Stop messages and execution time
        self._compute_results()

        return self.num_acquisitions

    def evaluate_objective(self):
        """
        Evaluates the objective
        """
        Y_new, _ = self.objective.evaluate(self.suggested_sample)
        return Y_new

    def _compute_next_evaluations(self):
        """
        Computes the location of the new evaluation (optimizes the acquisition in the standard case).
        """
        return self.evaluator.compute_batch()

    def _distance_last_evaluations(self, sample_size):
        """
        Computes the distance between the last two evaluations.
        """
        i = self.num_acquisitions
        last = self.X[i, :]
        previous = self.X[i - sample_size, :]  # in Batch mode only the first sample is non-random
        distance = np.linalg.norm(last - previous, ord=1)  # Manhattan distance
        return distance

    def _update_model(self):
        """
        Updates the model (when more than one observation is available) and saves the parameters (if available).
        """
        i = self.num_acquisitions
        if i % self.model_update_interval == 0:
            self.model.updateModel(self.X[:i], self.Y[:i])

    def _compute_results(self):
        """
        Computes the optimum and its value.
        """
        i = self.num_acquisitions
        Y = self.Y[:i]
        self.Y_best = best_value(Y)
        self.x_opt = self.X[np.argmin(Y), :]
        self.fx_opt = np.min(Y)

    def plot_convergence(self, filename=None):
        """
        Makes twp plots to evaluate the convergence of the model:
            plot 1: Iterations vs. distance between consecutive selected x's
            plot 2: Iterations vs. the mean of the current model in the selected sample.
        :param filename: name of the file where the plot is saved
        """
        i = self.Y_best.shape[0]
        return plot_convergence(self.X[:i], self.Y_best, filename)

    def get_evaluations(self):
        i = self.num_acquisitions
        return self.X[:i], self.Y[:i]

max_stars_repo_name: blindedjoy/Reservoir

max_issues_repo_name: blindedjoy/Reservoir

max_forks_repo_name: blindedjoy/Reservoir


text:
"""
This module converts objects into numpy array.
"""
import numpy as np
import torch
import six

def make_np(x):
    """
    Args:
      x: An instance of torch tensor or caffe blob name

    Returns:
        numpy.array: Numpy array
    """
    if isinstance(x, np.ndarray):
        return x
    if isinstance(x, six.string_types):  # Caffe2 will pass name of blob(s) to fetch
        return _prepare_caffe2(x)
    if np.isscalar(x):
        return np.array([x])
    if isinstance(x, torch.Tensor):
        return _prepare_pytorch(x)
    raise NotImplementedError(
        'Got {}, but numpy array, torch tensor, or caffe2 blob name are expected.'.format(type(x)))

def _prepare_pytorch(x):
    if isinstance(x, torch.autograd.Variable):
        x = x.data
    x = x.cpu().numpy()
    return x

def _prepare_caffe2(x):
    from caffe2.python import workspace
    x = workspace.FetchBlob(x)
    return x


text:
# coding: utf-8
# 2020/5/13 @ tongshiwei
from copy import deepcopy
import numpy as np
from collections import namedtuple
from EduSim.Envs.MBS.utils import hlr
from EduSim.Envs.meta import MetaLearner, MetaLearningModel
from .MBSLearner import MBSLearnerGroup

HLRState = namedtuple("HLRState", ["features", "latest_review_ts"])

class HLRLearningModel(MetaLearningModel):
    def step(self, state: HLRState, learning_item_id, outcome, timestamp, *args, **kwargs):
        state.features[learning_item_id][0] += 1
        state.features[learning_item_id][1 if outcome == 1 else 2] += 1
        state.latest_review_ts[learning_item_id] = timestamp

class HLRLearner(MetaLearner):
    def __init__(self, features, latest_review_ts, feature_coefficients):
        super(HLRLearner, self).__init__()
        self._state = HLRState(features, latest_review_ts)
        self._learning_model = HLRLearningModel()
        self._feature_coefficients = feature_coefficients

    @property
    def state(self):
        return HLRState([f[:3] for f in self._state.features], self._state.latest_review_ts)

    def learn(self, learning_item, timestamp, *args, **kwargs):
        self._learning_model.step(self._state, learning_item.id, self.response(learning_item, timestamp), timestamp)

    def response(self, test_item, timestamp, *args, **kwargs) -> ...:
        return hlr(
            timestamp - self._state.latest_review_ts[test_item.id],
            self._feature_coefficients,
            self._state.features[test_item.id],
        )

class HLRLearnerGroup(MBSLearnerGroup):
    def __init__(self, n_items, seed=None):
        self.n_items = n_items
        super(HLRLearnerGroup, self).__init__(seed)
        # n_attempts, n_correct, n_incorrect
        _features = np.zeros((self.n_items, 3))
        self._log_linear_features = self.sample_log_linear_feature(self.n_items)
        self._log_linear_features_coefficients = self.sample_log_linear_feature_coefficients(self.n_items)
        self._init_latest_review_ts = self.sample_init_review_time(self.n_items, sample_type="normal")

    def __next__(self):
        return HLRLearner(
            features=self._log_linear_features,
            latest_review_ts=deepcopy(self._init_latest_review_ts),
            feature_coefficients=self._log_linear_features_coefficients
        )

max_stars_repo_name: bigdata-ustc/EduSim

max_forks_repo_name: bigdata-ustc/EduSim


text:
import typing as ty
import numpy as np
from numpy.typing import NDArray
from dataclasses import dataclass, astuple

@dataclass(frozen=True)
class Dataset:
    train_x: NDArray[np.float_]
    train_y: NDArray[np.float_]
    test_x: NDArray[np.float_]
    test_y: NDArray[np.float_]

class ModelPredictions(ty.NamedTuple):
    y: NDArray[np.float_]
    std: NDArray[np.float_]
    # upper: NDArray[np.float_]

max_stars_repo_name: sutoiku/autostat

max_issues_repo_name: sutoiku/autostat

max_forks_repo_name: sutoiku/autostat

reason: import numpy,from numpy


text:
#!/usr/bin/env python
"""
utilities for XRF display
"""
import copy
from functools import partial
import numpy as np
import wx
import wx.lib.agw.pycollapsiblepane as CP
import wx.lib.agw.flatnotebook as flat_nb
import wx.lib.scrolledpanel as scrolled

from wxutils import (SimpleText, FloatCtrl, Choice, Font, pack, Button,
                     Check, HLine, GridPanel, RowPanel, CEN, LEFT, RIGHT)

from larch import Group, Parameter, Minimizer, fitting
from larch.larchlib import Empty
from larch_plugins.math import index_of, gaussian

from larch_plugins.xrf import (xrf_background, xrf_calib_fitrois,
                               xrf_calib_compute, xrf_calib_apply)

from larch_plugins.xray import material_mu, material_get
from larch_plugins.wx import ParameterPanel

try:
    from collections import OrderedDict
except ImportError:
    from larch.utils import OrderedDict

FNB_STYLE = flat_nb.FNB_NO_X_BUTTON|flat_nb.FNB_SMART_TABS|flat_nb.FNB_NO_NAV_BUTTONS

def read_filterdata(flist, _larch):
    """ read filters data"""
    materials = _larch.symtable.get_symbol('_xray._materials')
    out = OrderedDict()
    out['None'] = ('', 0)
    for name in flist:
        if name in materials:
            out[name]  = materials[name]
    return out

def xrf_resid(pars, peaks=None, mca=None, det=None, bgr=None,
              sig=None, filters=None, flyield=None, use_bgr=False,
              xray_en=30.0, emin=0.0, emax=30.0, **kws):

    sig_o = pars.sig_o.value
    sig_s = pars.sig_s.value
    sig_q = pars.sig_q.value
    model = mca.data * 0.0
    if use_bgr:
        model += mca.bgr
    for use, pcen, psig, pamp in peaks:
        amp = pamp.value
        cen = pcen.value
        sig = sig_o + cen * (sig_s + sig_q * cen)
        psig.value = sig
        if use: model += amp*gaussian(mca.energy, cen, sig)

    mca.model = model
    imin = index_of(mca.energy, emin)
    imax = index_of(mca.energy, emax)
    resid = (mca.data - model)
    # if det['use']:
    #     resid = resid / np.maximum(1.e-29, (1.0 - mca.det_atten))

    return resid[imin:imax]

class FitSpectraFrame(wx.Frame):
    """Frame for Spectral Analysis"""

    Filter_Lengths = ['microns', 'mm', 'cm']
    Filter_Materials = ['None', 'air', 'nitrogen', 'helium', 'argon', 'kapton',
                        'mylar', 'pmma', 'water', 'aluminum', 'silicon nitride',
                        'silicon', 'quartz', 'sapphire', 'diamond', 'graphite',
                        'boron nitride']

    Detector_Materials = ['Si', 'Ge']

    gsig_offset = 'sig_o'
    gsig_slope  = 'sig_s'
    gsig_quad   = 'sig_q'

    def __init__(self, parent, size=(675, 525)):
        self.parent = parent
        self.larch = parent.larch
        self.mca = parent.mca
        conf = parent.conf
        self.paramgroup = Group()

        if not hasattr(self.mca, 'init_calib'):
            xrf_calib_fitrois(self.mca, _larch=self.larch)

        wx.Frame.__init__(self, parent, -1, 'Fit XRF Spectra',
                          size=size, style=wx.DEFAULT_FRAME_STYLE)
        if not hasattr(self.parent, 'filters_data'):
            self.parent.filters_data = read_filterdata(self.Filter_Materials,
                                                       _larch=self.larch)

        self.wids = Empty()
        self.SetFont(Font(9))
        self.panels = {}
        self.nb = flat_nb.FlatNotebook(self, wx.ID_ANY, agwStyle=FNB_STYLE)
        self.nb.SetBackgroundColour('#FBFBF8')
        self.SetBackgroundColour('#F6F6F0')

        self.nb.AddPage(self.settings_page(), 'Fit & Background Settings')
        self.nb.AddPage(self.filters_page(),  'Filters and Attenuation')
        self.nb.AddPage(self.fitpeaks_page(), 'XRF Peaks')

        self.nb.SetSelection(0)

        sizer = wx.BoxSizer(wx.VERTICAL)
        sizer.Add(self.nb, 1, wx.ALL|wx.EXPAND)

        sizer.Add((5,5))
        sizer.Add(HLine(self, size=(675, 3)),  0, CEN|LEFT|wx.TOP|wx.GROW)
        sizer.Add((5,5))

        bpanel = RowPanel(self)
        bpanel.Add(Button(bpanel, 'Fit Peaks', action=self.onFitPeaks), 0, LEFT)
        bpanel.Add(Button(bpanel, 'Done', action=self.onClose), 0, LEFT)
        bpanel.pack()
        sizer.Add(bpanel, 0, CEN)
        pack(self, sizer)
        self.Show()
        self.Raise()

    def fitpeaks_page(self):
        "create row for filters parameters"
        mca = self.parent.mca
        self.wids.peaks = []

        p = GridPanel(self)

        p.AddManyText((' ROI Name', 'Fit?', 'Center', 'Sigma', 'Amplitude'),
                      style=CEN)
        p.Add(HLine(p, size=(600, 3)), dcol=5, newrow=True)
        offset, slope = self.mca.offset, self.mca.slope
        for iroi, roi in enumerate(self.mca.rois):
            try:
                cenval, ecen, fwhm, ampval, _x = self.mca.init_calib[roi.name]
            except KeyError:
                continue
            sigval = 0.4*fwhm
            mincen = offset + slope*(roi.left  - 4 * roi.bgr_width)
            maxcen = offset + slope*(roi.right + 4 * roi.bgr_width)

            pname = roi.name.replace(' ', '').lower()
            ampnam = '%s_amp' % pname
            cennam = '%s_cen' % pname
            signam = '%s_sig' % pname
            sigexpr = "%s + %s*%s +%s*%s**2" % (self.gsig_offset,
                                                self.gsig_slope, cennam,
                                                self.gsig_quad, cennam)

            p_amp = Parameter(value=ampval, vary=True,    min=0, name=ampnam)
            p_sig = Parameter(value=sigval, expr=sigexpr, min=0, name=signam)
            p_cen = Parameter(value=cenval, vary=False, name=cennam,
                              min=mincen, max=maxcen)

            setattr(self.paramgroup, ampnam, p_amp)
            setattr(self.paramgroup, cennam, p_cen)
            setattr(self.paramgroup, signam, p_sig)

            _use   = Check(p, label='use' , default=True)
            _cen   = ParameterPanel(p, p_cen, precision=3)
            _sig   = ParameterPanel(p, p_sig, precision=3)
            _amp   = ParameterPanel(p, p_amp, precision=2)

            self.wids.peaks.append((_use, _cen, _sig, _amp))

            p.AddText(' %s' % roi.name,  newrow=True, style=LEFT)
            p.Add(_use, style=wx.ALIGN_CENTER)
            p.Add(_cen)
            p.Add(_sig)
            p.Add(_amp)
        p.Add(HLine(p, size=(600, 3)), dcol=5, newrow=True)
        p.pack()
        return p

    def settings_page(self):
        "create fit and background settings"
        mca = self.parent.mca
        conf = self.parent.conf
        wids = self.wids

        width = getattr(mca, 'bgr_width',    4)
        compr = getattr(mca, 'bgr_compress', 2)
        expon = getattr(mca, 'bgr_exponent', 2)

        p = GridPanel(self, itemstyle=LEFT)
        wids.bgr_use = Check(p, label='Fit Background-Subtracted Spectra',
                             default=True)
        wids.bgr_width = FloatCtrl(p, value=width, minval=0, maxval=10,
                                   precision=1, size=(70, -1))
        wids.bgr_compress = Choice(p, choices=['1', '2', '4', '8', '16'],
                                   size=(70, -1), default=1)
        wids.bgr_exponent = Choice(p, choices=['2', '4', '6'],
                                   size=(70, -1), default=0)

        sopts = {'vary': True, 'precision': 5}
        sig_offset = Parameter(value=0.050, name=self.gsig_offset, vary=True)
        sig_slope  = Parameter(value=0.005, name=self.gsig_slope, vary=True)
        sig_quad   = Parameter(value=0.000, name=self.gsig_quad, vary=False)

        setattr(self.paramgroup, self.gsig_offset, sig_offset)
        setattr(self.paramgroup, self.gsig_slope, sig_slope)
        setattr(self.paramgroup, self.gsig_quad,  sig_quad)

        wids.sig_offset = ParameterPanel(p, sig_offset, vary=True, precision=5)
        wids.sig_slope  = ParameterPanel(p, sig_slope, vary=True, precision=5)
        wids.sig_quad   = ParameterPanel(p, sig_quad, vary=False, precision=5)

        wids.xray_en = FloatCtrl(p, value=20.0, size=(70, -1),
                                 minval=0, maxval=1000, precision=3)

        wids.fit_emin = FloatCtrl(p, value=conf.e_min, size=(70, -1),
                                  minval=0, maxval=1000, precision=3)
        wids.fit_emax = FloatCtrl(p, value=conf.e_max, size=(70, -1),
                                  minval=0, maxval=1000, precision=3)
        wids.flyield_use = Check(p,
            label='Account for Absorption, Fluorescence Efficiency')

        p.AddText(' General Settings ', colour='#880000', dcol=3)

        p.AddText(' X-ray Energy (keV): ', newrow=True)
        p.Add(wids.xray_en)
        p.AddText(' Min Energy (keV): ', newrow=True)
        p.Add(wids.fit_emin)
        p.AddText(' Max Energy (keV): ', newrow=False)
        p.Add(wids.fit_emax)

        wids.det_mat = Choice(p, choices=self.Detector_Materials,
                              size=(55, -1), default=0)
        wids.det_thk = FloatCtrl(p, value=0.40, size=(70, -1),
                                 minval=0, maxval=100, precision=3)
        wids.det_use = Check(p, label='Account for Detector Thickness')

        p.AddText(' Detector Material:  ', newrow=True)
        p.Add(wids.det_mat)
        p.AddText(' Thickness (mm): ', newrow=False)
        p.Add(wids.det_thk)
        p.Add(wids.det_use,     dcol=4, newrow=True)
        p.Add(wids.flyield_use, dcol=4, newrow=True)

        p.Add(HLine(p, size=(600, 3)), dcol=5, newrow=True)

        p.AddText(' Energy Dependence of Peak Width:',
                  colour='#880000', newrow=True, dcol=2)
        # p.AddText(' ', newrow=True)
        p.AddText(' sigma = offset + slope * Energy + quad * Energy^2 (keV)',
                  dcol=3)
        p.AddText(' Offset: ', newrow=True)
        p.Add(wids.sig_offset, dcol=3)
        p.AddText(' Slope: ', newrow=True)
        p.Add(wids.sig_slope, dcol=3)
        p.AddText(' Quad: ', newrow=True)
        p.Add(wids.sig_quad, dcol=3)
        p.Add(HLine(p, size=(600, 3)), dcol=5, newrow=True)

        p.AddText(" Background Parameters: ", colour='#880000', dcol=2,
                  newrow=True)
        p.Add(wids.bgr_use, dcol=2)

        p.AddText(" Exponent:", newrow=True)
        p.Add(wids.bgr_exponent)
        p.AddText(" Energy Width (keV): ", newrow=False)
        p.Add(wids.bgr_width)

        p.AddText(" Compression: ", newrow=True)
        p.Add(wids.bgr_compress)
        p.Add(Button(p, 'Show Background', size=(130, -1),
                     action=self.onShowBgr), dcol=2)

        p.Add(HLine(p, size=(600, 3)), dcol=5, newrow=True)
        p.pack()
        return p

    def onShowBgr(self, event=None):
        wids     = self.wids
        mca      = self.mca
        parent   = self.parent
        width    = wids.bgr_width.GetValue()
        compress = int(wids.bgr_compress.GetStringSelection())
        exponent = int(wids.bgr_exponent.GetStringSelection())
        xrf_background(energy=mca.energy, counts=mca.counts,
                       group=mca, width=width, compress=compress,
                       exponent=exponent, _larch=parent.larch)
        mca.bgr_width = width
        mca.bgr_compress = compress
        mca.bgr_exponent = exponent
        parent.plotmca(mca)
        parent.oplot(mca.energy, mca.bgr, label='background',
                     color=parent.conf.bgr_color, linewidth=1, style='--')

    def onFilterMaterial(self, evt=None, index=0):
        name = evt.GetString()
        form, den = self.parent.filters_data.get(name, ('', 0))
        self.wids.filters[index][1].SetValue(den)
        thick = self.wids.filters[index][2]
        if thick.wids.val.GetValue()  < 1.e-5:
            thick.wids.val.SetValue(0.0250)

    def onEditFilters(self, evt=None):
        print( 'on Edit Filters ',  evt)

    def filters_page(self):
        "create row for filters parameters"
        mca = self.parent.mca
        self.wids.filters = []

        p = GridPanel(self, itemstyle=LEFT)

        bx = Button(p, 'Customize Filter List', size=(150, -1),
                    action = self.onEditFilters)
        bx.Disable()
        p.AddManyText((' filter', 'material', 'density (gr/cm^3)',
                       'thickness (mm)'), style=CEN)
        p.Add(HLine(p, size=(600, 3)), dcol=5, newrow=True)
        for i in range(4):
            _mat = Choice(p, choices=self.Filter_Materials, default=0,
                          size=(125, -1),
                          action=partial(self.onFilterMaterial, index=i))
            _den = FloatCtrl(p, value=0, minval=0, maxval=30,
                             precision=4, size=(75, -1))

            pnam = 'filter%i_thickness' % (i+1)
            param = Parameter(value=0.0, vary=False, min=0, name=pnam)
            setattr(self.paramgroup, pnam, param)
            _len  = ParameterPanel(p, param, precision=4)

            self.wids.filters.append((_mat, _den, _len))
            p.AddText('  %i ' % (i+1), newrow=True)
            p.Add(_mat)
            p.Add(_den, style=wx.ALIGN_CENTER)
            p.Add(_len)
        p.Add(HLine(p, size=(600, 3)), dcol=5, newrow=True)
        p.AddText(' ', newrow=True)
        p.Add(bx, dcol=3)

        p.pack()
        return p

    def onFitPeaks(self, event=None):
        opts = {}
        filters, peaks = [], []
        sig, det, bgr = {}, {}, {}

        opts['flyield']  = self.wids.flyield_use.IsChecked()
        opts['xray_en']  = self.wids.xray_en.GetValue()
        opts['emin']     = self.wids.fit_emin.GetValue()
        opts['emax']     = self.wids.fit_emax.GetValue()

        det['use']       = self.wids.det_use.IsChecked()
        det['thickness'] = self.wids.det_thk.GetValue()
        det['material']  = self.wids.det_mat.GetStringSelection()

        bgr['use']       = self.wids.bgr_use.IsChecked()
        bgr['width']     = self.wids.bgr_width.GetValue()
        bgr['compress']  = int(self.wids.bgr_compress.GetStringSelection())
        bgr['exponent']  = int(self.wids.bgr_exponent.GetStringSelection())

        sig['offset']    = self.wids.sig_offset.param
        sig['slope']     = self.wids.sig_slope.param
        sig['quad']      = self.wids.sig_quad.param

        for k in self.wids.filters:
            f = (k[0].GetStringSelection(), k[1].GetValue(), k[2].param)
            filters.append(f)

        for k in self.wids.peaks:
            use = k[0].IsChecked()
            p = (k[0].IsChecked(), k[1].param, k[2].param, k[3].param)
            peaks.append(p)
            if not use:
                k[1].param.vary = False
                k[2].param.vary = False
                k[3].param.vary = False

        opts['det'] = det
        opts['bgr'] = bgr
        opts['sig'] = sig
        opts['filters'] = filters
        opts['peaks'] = peaks

        mca    = self.mca
        mca.data = mca.counts*1.0
        energy = mca.energy
        _larch = self.parent.larch
        if bgr['use']:
            bgr.pop('use')
            xrf_background(energy=mca.energy, counts=mca.counts,
                           group=mca, _larch=_larch, **bgr)
            opts['use_bgr']=True
        opts['mca'] = mca
        if det['use']:
            mu = material_mu(det['material'], energy*1000.0,
                             _larch=_larch)/10.0
            t = det['thickness']
            mca.det_atten = np.exp(-t*mu)

        fit = Minimizer(xrf_resid, self.paramgroup, toler=1.e-4,
                        _larch=_larch, fcn_kws = opts)
        fit.leastsq()
        parent = self.parent
        parent.oplot(mca.energy, mca.model,
                     label='fit', style='solid',
                     color='#DD33DD')

        print( fitting.fit_report(self.paramgroup, _larch=_larch))

        # filters:
        #  75 microns kapton, etc
        #
        # form, nominal_density = material_get(material)
        # mu = material_mu(material, energy*1000.0, _larch=_larch)/10.0
        # mu = mu * nominal_density/user_density
        # scale = exp(-thickness*mu)

    def onClose(self, event=None):
        self.Destroy()

max_stars_repo_name: bruceravel/xraylarch

max_issues_repo_name: bruceravel/xraylarch

max_forks_repo_name: bruceravel/xraylarch


text:
'''Various Estimators for Recommender related values.'''

import numpy as np
from sklearn.base import BaseEstimator, ClassifierMixin, RegressorMixin, MultiOutputMixin
from sklearn.multioutput import MultiOutputClassifier

class FunkSVDEstimator(BaseEstimator, RegressorMixin):
  '''Uses FunkSVD to interact with .

  Args:
    epochs (int): Number of epochs for the fit phase
    lr (float): Learning Rate used
    latent_features (int): number of latent features to learn through FunkSVD
    early_stop (bool): Defines if early stopping is enabled
    early_stop_thres (float): Threshold for early stopping (default: `0.0001`)
    debug (bool): Defines if debug output for training epochs should be given
  '''
  def __init__(self, epochs=25, lr=0.005, latent_features=15, early_stop=False, early_stop_thres=0.0001, debug=False):
    # set relevant values
    self.lr = lr
    self.epochs = epochs
    self.latent = latent_features
    self.early_stop = early_stop
    self.early_thres = early_stop_thres
    self.debug = debug
    # generate random start matrix
    self.user_mat = None
    self.item_mat = None
    self.user_idxs = None
    self.item_idxs = None

  def fit(self, X, y=None):
    '''Fits the SVD matricies to the given values.

    Args:
      X (np.array): 2D Numpy Array that contains the user-item matrix - Note that the column and row indices of the array have to be set to meaningful values
    '''
    # TODO: error checks
    if not isinstance(X, pd.DataFrame):
      print("Warning: X value to fit is not a pandas Dataframe, might not be able to retrieve accurate ids for predict function...")
      X = pd.DataFrame(X)

    # init with random data
    self.user_mat = np.random.rand(X.shape[0], self.latent)
    self.item_mat = np.random.rand(self.latent, X.shape[1])
    accum_sse = np.Inf

    # store indices for later retrival
    self.user_idxs = np.array(X.index)
    self.item_idxs = np.array(X.columns)

    # check for debug output
    if self.debug: print("Iterations | Mean Squared Error ")

    # iterate through all epochs
    for ep in range(self.epochs):
      # setup vals
      old_sse = accum_sse
      accum_sse = 0
      num_items = 0

      # For each user-movie pair
      for u_idx in range(X.shape[0]):
        for m_idx in range(X.shape[1]):
          # if the rating exists
          if not np.isnan(X.iloc[u_idx, m_idx]):
            # compute the error as the actual minus the dot product of the user and movie latent features
            act = X.iloc[u_idx, m_idx]
            pu = self.user_mat[u_idx, :]
            pm = self.item_mat[:, m_idx]
            pred = np.sum(np.dot(pu, pm))
            accum_sse += (pred - act) ** 2
            num_items += 1

            # update the values in each matrix in the direction of the gradient
            pu_new = pu + self.lr * 2 * (act-pred) * pm
            pm_new = pm + self.lr * 2 * (act-pred) * pu_new
            self.user_mat[u_idx, :]  = pu_new
            self.item_mat[:, m_idx] = pm_new

      # debug output
      accum_sse = accum_sse / max([num_items, 1])
      if self.debug: print("{:10} | {:.6f}".format(ep, accum_sse))

      # check for early stopping
      if self.early_stop == True and old_sse - accum_sse < self.early_thres:
        break

    return self

  def predict(self, X):
    '''Predicts the ratings for the given parings.

    Args:
      X (np.array): Array of data items to predict. Either single value tuple in form `(user_id, item_id)` or list of tuples

    Returns:
      Array of predicted values
    '''
    # check for init
    if self.user_mat is None or self.item_mat is None:
      raise RuntimeError("FunkSVDEstimator has not been fitted!")

    # retrieve the relevant items to predict (if possible)
    if isinstance(X, tuple) or len(X.shape) == 1:
      X = [X]
    # TODO: additional error / type checking

    # iterate through predicitons
    res = []
    for row in X:
      # retrieve data
      user_id = np.where(self.user_idxs == row[0])[0]
      item_id = np.where(self.item_idxs == row[1])[0]
      # cannot predict if not in training
      if len(user_id) == 0 or len(item_id) == 0:
        res.append(np.nan)
      # predict value
      pred = np.dot(self.user_mat[user_id[0], :], self.item_mat[:, item_id[0]])
      res.append(pred)

    return np.array(res)

  def score(self, X, y=None):
    '''Provides the RMSE score for the given items.

    Args:
      X (np.array): set of validation data-points to check against - should be in format `user | item | value`

    Returns:
      RMSE score for all predictable items as a float
    '''
    # setup values
    rmse = 0
    n_items = 0

    # iterate through all values
    for row in X.iterrows():
      # cannot calc if no rating
      if np.isnan(row[2]): continue

      # retrieve data
      user_id = np.where(self.user_idxs == row[0])[0]
      item_id = np.where(self.item_idxs == row[1])[0]

      # cannot predict if not in training data
      if len(user_id) == 0 or len(item_id) == 0: continue

      # predict data
      pred = np.dot(self.user_mat[user_id[0], :], self.item_mat[:, item_id[0]])
      rmse += (act - pred) ** 2
      n_items += 1

    # check if
    if n_items == 0: return np.Inf

    # calculate and return
    return np.sqrt(rmse/n_items)

# TODO: similar to FunkSVD Estimator create neural network estimator that trains a TF model?
class NeuralValueEstimator(BaseEstimator, RegressorMixin):
  '''Value Estimator that uses a neural network to train the prediction of relevant input items.'''
  # TODO: define which input features to use?
  pass


text:
# Copyright (c) 2021, NVIDIA CORPORATION.  All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from math import ceil
from time import perf_counter
from typing import List

import numpy as np
import torch.nn as nn
from tqdm import tqdm

from nemo.collections.nlp.data.text_normalization import TextNormalizationTestDataset, constants
from nemo.collections.nlp.data.text_normalization.utils import input_preprocessing, post_process_punct
from nemo.collections.nlp.models.duplex_text_normalization.utils import get_formatted_string
from nemo.utils import logging

__all__ = ['DuplexTextNormalizationModel', 'post_process_punct']

class DuplexTextNormalizationModel(nn.Module):
    """
    DuplexTextNormalizationModel is a wrapper class that can be used to
    encapsulate a trained tagger and a trained decoder. The class is intended
    to be used for inference only (e.g., for evaluation).
    """

    def __init__(self, tagger, decoder, lang):
        super(DuplexTextNormalizationModel, self).__init__()

        self.tagger = tagger
        self.decoder = decoder
        self.lang = lang

    def evaluate(
        self, dataset: TextNormalizationTestDataset, batch_size: int, errors_log_fp: str, verbose: bool = True
    ):
        """ Function for evaluating the performance of the model on a dataset

        Args:
            dataset: The dataset to be used for evaluation.
            batch_size: Batch size to use during inference. You can set it to be 1
                (no batching) if you want to measure the running time of the model
                per individual example (assuming requests are coming to the model one-by-one).
            errors_log_fp: Path to the file for logging the errors
            verbose: if true prints and logs various evaluation results

        Returns:
            results: A Dict containing the evaluation results (e.g., accuracy, running time)
        """
        results = {}
        error_f = open(errors_log_fp, 'w+')

        # Apply the model on the dataset
        (
            all_run_times,
            all_dirs,
            all_inputs,
            all_targets,
            all_classes,
            all_nb_spans,
            all_span_starts,
            all_span_ends,
            all_output_spans,
        ) = ([], [], [], [], [], [], [], [], [])
        all_tag_preds, all_final_preds = [], []
        nb_iters = int(ceil(len(dataset) / batch_size))
        for i in tqdm(range(nb_iters)):
            start_idx = i * batch_size
            end_idx = (i + 1) * batch_size
            batch_insts = dataset[start_idx:end_idx]
            (
                batch_dirs,
                batch_inputs,
                batch_targets,
                batch_classes,
                batch_nb_spans,
                batch_span_starts,
                batch_span_ends,
            ) = zip(*batch_insts)
            # Inference and Running Time Measurement
            batch_start_time = perf_counter()

            batch_tag_preds, batch_output_spans, batch_final_preds = self._infer(
                batch_inputs, batch_dirs, processed=True
            )

            batch_run_time = (perf_counter() - batch_start_time) * 1000  # milliseconds
            all_run_times.append(batch_run_time)
            # Update all_dirs, all_inputs, all_tag_preds, all_final_preds and all_targets
            all_dirs.extend(batch_dirs)
            all_inputs.extend(batch_inputs)
            all_tag_preds.extend(batch_tag_preds)
            all_final_preds.extend(batch_final_preds)
            all_targets.extend(batch_targets)
            all_classes.extend(batch_classes)
            all_nb_spans.extend(batch_nb_spans)
            all_span_starts.extend(batch_span_starts)
            all_span_ends.extend(batch_span_ends)
            all_output_spans.extend(batch_output_spans)

        # Metrics
        tn_error_ctx, itn_error_ctx = 0, 0
        for direction in constants.INST_DIRECTIONS:
            (
                cur_dirs,
                cur_inputs,
                cur_tag_preds,
                cur_final_preds,
                cur_targets,
                cur_classes,
                cur_nb_spans,
                cur_span_starts,
                cur_span_ends,
                cur_output_spans,
            ) = ([], [], [], [], [], [], [], [], [], [])
            for dir, _input, tag_pred, final_pred, target, cls, nb_spans, span_starts, span_ends, output_spans in zip(
                all_dirs,
                all_inputs,
                all_tag_preds,
                all_final_preds,
                all_targets,
                all_classes,
                all_nb_spans,
                all_span_starts,
                all_span_ends,
                all_output_spans,
            ):
                if dir == direction:
                    cur_dirs.append(dir)
                    cur_inputs.append(_input)
                    cur_tag_preds.append(tag_pred)
                    cur_final_preds.append(final_pred)
                    cur_targets.append(target)
                    cur_classes.append(cls)
                    cur_nb_spans.append(nb_spans)
                    cur_span_starts.append(span_starts)
                    cur_span_ends.append(span_ends)
                    cur_output_spans.append(output_spans)
            nb_instances = len(cur_final_preds)
            cur_targets_sent = [" ".join(x) for x in cur_targets]

            sent_accuracy = TextNormalizationTestDataset.compute_sent_accuracy(
                cur_final_preds, cur_targets_sent, cur_dirs
            )

            class_accuracy = TextNormalizationTestDataset.compute_class_accuracy(
                [x.split() for x in cur_inputs],
                cur_targets,
                cur_tag_preds,
                cur_dirs,
                cur_output_spans,
                cur_classes,
                cur_nb_spans,
                cur_span_ends,
            )
            if verbose:
                logging.info(f'\n============ Direction {direction} ============')
                logging.info(f'Sentence Accuracy: {sent_accuracy}')
                logging.info(f'nb_instances: {nb_instances}')
                if not isinstance(class_accuracy, str):
                    log_class_accuracies = ""
                    for key, value in class_accuracy.items():
                        log_class_accuracies += f"\n\t{key}:\t{value[0]}\t{value[1]}/{value[2]}"
                else:
                    log_class_accuracies = class_accuracy
                logging.info(f'class accuracies: {log_class_accuracies}')
            # Update results
            results[direction] = {
                'sent_accuracy': sent_accuracy,
                'nb_instances': nb_instances,
                "class_accuracy": log_class_accuracies,
            }
            # Write errors to log file
            for _input, tag_pred, final_pred, target, classes in zip(
                cur_inputs, cur_tag_preds, cur_final_preds, cur_targets_sent, cur_classes
            ):
                if not TextNormalizationTestDataset.is_same(final_pred, target, direction):
                    if direction == constants.INST_BACKWARD:
                        error_f.write('Backward Problem (ITN)\n')
                        itn_error_ctx += 1
                    elif direction == constants.INST_FORWARD:
                        error_f.write('Forward Problem (TN)\n')
                        tn_error_ctx += 1

                    formatted_input_str = get_formatted_string(self.decoder.processor.tokenize(_input).split())
                    formatted_tag_pred_str = get_formatted_string(tag_pred)
                    class_str = " ".join(classes)
                    error_f.write(f'Original Input : {_input}\n')
                    error_f.write(f'Input          : {formatted_input_str}\n')
                    error_f.write(f'Predicted Tags : {formatted_tag_pred_str}\n')
                    error_f.write(f'Ground Classes : {class_str}\n')
                    error_f.write(f'Predicted Str  : {final_pred}\n')
                    error_f.write(f'Ground-Truth   : {target}\n')
                    error_f.write('\n')
            results['itn_error_ctx'] = itn_error_ctx
            results['tn_error_ctx'] = tn_error_ctx

        # Running Time
        avg_running_time = np.average(all_run_times) / batch_size  # in ms
        if verbose:
            logging.info(f'Average running time (normalized by batch size): {avg_running_time} ms')
        results['running_time'] = avg_running_time

        # Close log file
        error_f.close()
        logging.info(f'Errors are saved at {errors_log_fp}.')
        return results

    # Functions for inference
    def _infer(self, sents: List[str], inst_directions: List[str], processed=False):
        """
        Main function for Inference

        If the 'joint' mode is used, "sents" will include both spoken and written forms on each input sentence,
        and "inst_directions" will include both constants.INST_BACKWARD and constants.INST_FORWARD

        Args:
            sents: A list of input texts.
            inst_directions: A list of str where each str indicates the direction of the corresponding instance \
                (i.e., constants.INST_BACKWARD for ITN or constants.INST_FORWARD for TN).
            processed: Set to True when used with TextNormalizationTestDataset, the data is already tokenized with moses,
                repetitive moses tokenization could lead to the number of tokens and class span mismatch

        Returns:
            tag_preds: A list of lists where the inner list contains the tag predictions from the tagger for each word in the input text.
            output_spans: A list of lists where each list contains the decoded semiotic spans from the decoder for an input text.
            final_outputs: A list of str where each str is the final output text for an input text.
        """
        original_sents = [s for s in sents]
        # Separate into words
        if not processed:
            sents = [input_preprocessing(x, lang=self.lang) for x in sents]
            sents = [self.decoder.processor.tokenize(x).split() for x in sents]
        else:
            sents = [x.split() for x in sents]

        # Tagging
        # span_ends included, returns index wrt to words in input without auxiliary words
        tag_preds, nb_spans, span_starts, span_ends = self.tagger._infer(sents, inst_directions)
        output_spans = self.decoder._infer(sents, nb_spans, span_starts, span_ends, inst_directions)

        # Prepare final outputs
        final_outputs = []
        for ix, (sent, tags) in enumerate(zip(sents, tag_preds)):
            try:
                cur_words, jx, span_idx = [], 0, 0
                cur_spans = output_spans[ix]
                while jx < len(sent):
                    tag, word = tags[jx], sent[jx]
                    if constants.SAME_TAG in tag:
                        cur_words.append(word)
                        jx += 1
                    else:
                        jx += 1
                        cur_words.append(cur_spans[span_idx])
                        span_idx += 1
                        while jx < len(sent) and tags[jx] == constants.I_PREFIX + constants.TRANSFORM_TAG:
                            jx += 1

                if processed:
                    # for Class-based evaluation, don't apply Moses detokenization
                    cur_output_str = " ".join(cur_words)
                else:
                    # detokenize the output with Moses and fix punctuation marks to match the input
                    # for interactive inference or inference from a file
                    cur_output_str = self.decoder.processor.detokenize(cur_words)
                    cur_output_str = post_process_punct(input=original_sents[ix], normalized_text=cur_output_str)
                final_outputs.append(cur_output_str)
            except IndexError:
                logging.warning(f"Input sent is too long and will be skipped - {original_sents[ix]}")
                final_outputs.append(original_sents[ix])
        return tag_preds, output_spans, final_outputs

max_stars_repo_name: hamjam/NeMo

max_issues_repo_name: hamjam/NeMo

max_forks_repo_name: hamjam/NeMo


text:
import numpy as np

class MCTDH:

    def __init__(self):
        """
        """

max_stars_repo_name: addschile/pymctdh

max_issues_repo_name: addschile/pymctdh

max_forks_repo_name: addschile/pymctdh


text:
import urllib.request
import html2text
import numpy as np
from scipy.optimize import curve_fit
import os

## Used by push_recipe to optimize words reading & words matching

try:
  import matplotlib as matplot
  matplot.use('Agg')
  import matplotlib.pyplot as mpl
  plottingEnabled = True
except:
  print("Matplotlib not detected, plotting disabled")
  plottingEnabled = False

def normpdf(x, a, x0, sigma):
    return a*np.exp(-(x-x0)**2/(2*sigma**2))

def count_occurences(context, sentence):
  sentence = sentence.lower()
  occurences = 0
  words_in_sentence = len(sentence.split())
  for word in context['context_words']:
    if words_in_sentence < context['hasFewerWordsThan'] and words_in_sentence > context['hasMoreWordsThan']:
      occurences = occurences + sentence.count(word)
      # If spaces are important, use: occurences = occurences + sum(1 for _ in re.finditer(r'\b%s\b' % re.escape(word), sentence))
  if occurences > 0 and context['hasSpecial'] in sentence:
    occurences = occurences *2
  for word in context['bad_context_words']:
    if words_in_sentence < context['hasFewerWordsThan'] and words_in_sentence > context['hasMoreWordsThan']:
      occurences = occurences - sentence.count(word)
      # If spaces are important, use: occurences = occurences + sum(1 for _ in re.finditer(r'\b%s\b' % re.escape(word), sentence))
  return occurences

def get_url_markdown(baseurl):
  opener = urllib.request.build_opener()
  opener.addheaders = [('User-agent', 'Mozilla/5.0 (Windows NT 6.3; rv:36.0) Gecko/20100101 Firefox/36.0')]
  j = opener.open(baseurl)
  data = j.read()

  h = html2text.HTML2Text()
  h.ignore_links = True
  h.ignore_images = True
  h.body_width = 10000
  return h.handle(data.decode('utf8'))

def get_occurrences(contexts,text):
  o_array = {}
  x = np.array([0])
  for context in contexts:
    o_array[context] = np.array([0])

  line_number = 0
  for line in text.splitlines():
    line_number = line_number + 1
    x = np.append(x,[line_number])
    #print str(line_number) + ":\t",
    for i in contexts:
      num_occurences = count_occurences(contexts[i],line.strip())
      #print str(num_occurences) + "\t",
      o_array[i] = np.append(o_array[i],[num_occurences])
    #print line
  return o_array

def calculate_context_peaks(contexts,o_array):
  if plottingEnabled:
    fig = mpl.figure()
  contextNum = 0
  o_fits = {}
  for context in contexts:
    yi=np.array([])
    xi=np.array([])
    a=o_array[context]
    for i in np.arange(1,len(a)-5,5):
        yi = np.append(yi,np.sum(a[i:i+5]))
        xi=np.append(xi,i+3)
    for i in np.arange(2,len(a)-5,5):
        yi = np.append(yi,np.sum(a[i:i+5]))
        xi=np.append(xi,i+3)
    for i in np.arange(3,len(a)-5,5):
        yi = np.append(yi,np.sum(a[i:i+5]))
        xi=np.append(xi,i+3)
    for i in np.arange(4,len(a)-5,5):
        yi = np.append(yi,np.sum(a[i:i+5]))
        xi=np.append(xi,i+3)
    for i in np.arange(5,len(a)-5,5):
        yi = np.append(yi,np.sum(a[i:i+5]))
        xi=np.append(xi,i+3)

    #popt, pcov = curve_fit(normpdf, x, y])
    maxIndex = int(xi[np.argmax(yi)])
    bestArea = 0

#    for i in range(1,o_array[context].size):
    for i in range(maxIndex-3,maxIndex+3):
      try:
        popt, pcov = curve_fit(normpdf, xi, yi, p0=[np.max(yi),i,1])
        if popt[0]*0.5*popt[2] > bestArea and popt[2]<15:
          bestArea = popt[0]*0.5*popt[2]
          best_popt = popt
          best_pcov = pcov
      except:
        pass
    popt = best_popt
    o_fits[context] = popt
    pcov = best_pcov
    if plottingEnabled:
      ym = normpdf(xi, popt[0], popt[1], popt[2])
      ax = fig.add_subplot(100*len(contexts)+11+contextNum)
      ax.scatter(xi, yi,label=context)
      ax.plot(xi, ym, c='r', label='Best fit')
      mpl.title(context)
      mpl.ylabel('Contextual density')
    contextNum = contextNum + 1
  if plottingEnabled:
    mpl.xlabel('Line number')
    fig.savefig('ingredients.png')
    mpl.close()
  return o_fits

def findTitle(text):
  title ="Unknown recipe"
  numPounds = 100
  for line in text.split('\n'):
    if line.count('#')<numPounds:
      numPounds = line.count('#')
      title = line.replace('#','').strip()
  return title

def get_snippets(contexts,source):
  if "http" in source:
    #print("Getting url " + source + "...")
    text = get_url_markdown(source)
  elif ".md" in source:
    if "bz2" in source:
      os.system('bzip2 -d ' + source)
      os.system('cp ' + source[0:-4] + ' ./source.tmp')
      os.system('bzip2 ' + source[0:-4])
    else:
      os.system('cp ' + source + ' ./source.tmp')
    text = ""
    with open('./source.tmp','rb') as f:
      for line in f:
        text = text + line.strip() + "\n"
    os.system('rm ./source.tmp')
  numberLines = 0
  for line in text.splitlines():
    numberLines = numberLines + 1
  totalLines = numberLines

  for context in contexts:
    newText = ""
    numberLines = 0
    for line in text.splitlines():
      newText = newText + line.strip() + "\n"
      numberLines = numberLines + 1
      if totalLines > 100 and numberLines > totalLines*float(contexts[context]['topProportion']):
        break
    contexts[context]["text"] = newText
  print("Getting number occurrences in each line...")
  o_array = get_occurrences(contexts,contexts[context]["text"])
  print("Curve fitting on single Gaussian...")
  o_fits = calculate_context_peaks(contexts,o_array)
  print("Grabbing snippets...")
  o_snippet = {}
  o_snippet['title'] = findTitle(text)
  for context in o_fits:
    o_snippet[context] = ""

  for context in o_fits:
    line_number = 0
    for line in contexts[context]['text'].splitlines():
      line_number = line_number + 1
      if line_number >=   o_fits[context][1]-round(2*o_fits[context][2]) and line_number <= o_fits[context][1]+round(2*o_fits[context][2]):
        if len(line)>1 and "##" not in line and len(line.split())<contexts[context]['ignoreIfWordsExceed'] and len(line.split())>contexts[context]['ignoreIfWordsLessThan']:
          o_snippet[context] = o_snippet[context] + line + "\n"

  return (o_snippet,o_fits,o_array)

#images_all = get_image("http://www.marthastewart.com/344840/soft-and-chewy-chocolate-chip-cookies")
##for img in images_all:
##  print img
##URL = "https://assets.marthastewart.com/styles/wmax-750/d49/soft-chewy-chocolate-chip-cookies-01-7004023-0819/soft-chewy-chocolate-chip-cookies-01-7004023-0819_horiz.jpg?itok=PPg28p8u"
##file = cStringIO.StringIO(urllib.urlopen(URL).read())
##im=Image.open(file)
##width, height = im.size
##print width
##print height

reason: import numpy,from scipy


text:
"""
用于stage1、stage4和stage5的关键词生成和相关信息提取
"""

import numpy as np
import matplotlib.pyplot as plt
import math
import os
import re
import Date
import time
import json
import jieba
import stage
import splitWeibo
from scipy.optimize import curve_fit
from scipy.stats import chi2_contingency
from scipy.stats import kstest
from sklearn.metrics import mean_squared_error, r2_score

stageNo = 6
path = stage.stage[stageNo]['path']
beginDate = stage.stage[stageNo]['beginDate']
print(beginDate)
endDate = stage.stage[stageNo]['endDate']
print(endDate)
date = '1970-01-01-'

keywords = []
scoreList = []
starList = []
forwardList = []
commentList = []
allWeibo = []
weiboNumByDate = []
COVWeiboNumList = []
ka = []
curveParam = []
ksRes = []
R2 = ''
MSE = ''
RMSE = ''

# textrank = analyse.textrank

def run():
    plt.rcParams['font.sans-serif'] = ['SimHei']
    keywords = keywordsInit()
    print('Begin')
    # 根据微博正文给所有微博打分
    date = beginDate
    while Date.dateCmp(date, endDate) != 1:
        getWeiboScores(date, keywords)
        readScoreWords(date)
        print(date + 'Finished')
        date = Date.getNextDate(date)
    global scoreList
    numOfWeibo = len(scoreList)

    limit = fitData()  # 第二、第三阶段fit出来都很完美，第一阶段fit不出来的
    drawPic()

    COVWeiboList = [weibo for weibo in allWeibo if weibo['疫情相关度'] > limit]
    for item in COVWeiboList:
        item['是否与疫情相关'] = True
    COVWeiboNum = len(COVWeiboList)

    print(weiboNumByDate)
    fp = open(path + '/' + 'stageInfo.json', mode='w', encoding='utf-8')
    stageInfo = {'微博数量': numOfWeibo, '疫情相关微博数量': COVWeiboNum, '卡方检验': str(ka), '拟合函数参数': str(curveParam),
                 'K-S': str(ksRes), 'MSE': MSE, 'RMSE': RMSE, 'R2': R2}
    # stage['疫情信息热度'] = 0
    outWeiboInfo = json.dumps(stageInfo, indent=4, separators=(',', ':'), ensure_ascii=False)
    fp.write(outWeiboInfo)
    fp.flush()

    fp = open(path + '/' + 'stageCOVWeibo.json', mode='w', encoding='utf-8')
    outCOVWeibo = json.dumps(COVWeiboList, indent=4, separators=(',', ':'), ensure_ascii=False)
    fp.write(outCOVWeibo)
    fp.flush()

    fp = open(path + '/' + 'stageCOVWeiboByImportance.json', mode='w', encoding='utf-8')
    COVWeiboList = sorted(COVWeiboList, key=lambda x: x['疫情相关度'], reverse=True)
    time.sleep(3)
    print('stageCOVWeiboByImportance.json')
    outCOVWeiboByImportance = json.dumps(COVWeiboList, indent=4, separators=(',', ':'), ensure_ascii=False)
    fp.write(outCOVWeiboByImportance)
    fp.flush()

    splitWeibo()

def getWeiboScores(date, keywords):
    i = 0
    fp = open(path + '/' + date + '/' + date + 'blog.json', mode='r', encoding='utf-8')
    curDateWeibo = json.load(fp)
    weiboNumByDate.append(len(curDateWeibo))
    for weibo in curDateWeibo:
        content = weibo['微博内容']
        # print(content)
        contentWordList = jieba.cut(content)
        score = 0
        keywordsWithoutWeight = [x[0] for x in keywords]
        for word in contentWordList:
            if word in keywordsWithoutWeight:
                score += keywords[keywordsWithoutWeight.index(word)][1]
        i += 1
        weibo['疫情相关度'] = score
        weibo['是否与疫情相关'] = False
        # print(score)
        scoreList.append(score)
        forwardList.append(weibo['转发数'])
        starList.append(weibo['点赞数'])
        commentList.append(weibo['评论数'])
        weibo['评论'] = weibo['评论'][0:math.floor(len(weibo['评论']) / 5)]
        # if i > 2:
        #     break
        allWeibo.append(weibo)
    fp = open(path + '/' + date + '/' + date + 'blog-Scored.json', mode='w', encoding='utf-8')
    outCurDateWeibo = json.dumps(curDateWeibo, indent=4, separators=(',', ':'), ensure_ascii=False)
    fp.write(outCurDateWeibo)

    # 生成当日微博基础信息，可能后面有增补
    fp = open(path + '/' + date + '/' + date + 'blogInfo.json', mode='w', encoding='utf-8')
    curDateWeiboInfo = {}
    curDateWeiboInfo['微博数量'] = len(curDateWeibo)
    curDateWeiboInfo['疫情相关微博数量'] = 0
    # curDateWeiboInfo['疫情信息热度'] = 0
    outWeiboInfo = json.dumps(curDateWeiboInfo, indent=4, separators=(',', ':'), ensure_ascii=False)
    fp.write(outWeiboInfo)
    fp.flush()

def readScoreWords(date):
    i = 0
    fp = open(path + '/' + date + '/' + date + 'blog-Scored.json', mode='r', encoding='utf-8')
    curDateWeibo = json.load(fp)
    weiboNumByDate.append(len(curDateWeibo))
    allWeiboNeedReload = False
    if allWeibo == {}:
        allWeiboNeedReload = True
    for weibo in curDateWeibo:
        content = weibo['微博内容']
        scoreList.append(weibo['疫情相关度'])
        starList.append(weibo['点赞数'])
        forwardList.append(weibo['转发数'])
        commentList.append(weibo['评论数'])
        if allWeiboNeedReload:
            allWeibo.append(weibo)
    weiboNumByDate.append(len(curDateWeibo))
    fp = open(path + '/' + date + '/' + date + 'blogInfo.json', mode='r', encoding='utf-8')
    curDateWeiboInfo = json.load(fp)
    # COVWeiboNumList.append(curDateWeiboInfo['疫情相关微博数量'])
    # print('当天疫情相关微博数量' + str(curDateWeiboInfo['疫情相关微博数量']))

def fitData():
    xplot = np.arange(1, len(scoreList) + 1)
    scoreListN = np.array(sorted(scoreList))
    popt, pcov = curve_fit(exponent, xplot, scoreListN)
    plt.plot(xplot, sorted(scoreList), 'b+:', label='原始相关度')
    fitedData = exponent(xplot, *popt)

    limit1 = np.percentile(fitedData, stage.stage[stageNo]['ratio'])
    fitedLine = [exponent(x, popt[0], popt[1], popt[2], popt[3]) for x in xplot]
    plt.axhline(y=limit1, color="purple")
    plt.rcParams['font.sans-serif'] = ['SimHei']
    plt.plot(xplot, fitedData, linestyle="-", marker="", color='r', label='拟合曲线')
    plt.title('阶段' + str(stageNo) + '疫情相关度与拟合相关度结果')
    plt.legend()
    plt.savefig(path + '/SaveTest_Fit.png', dpi=300)

    # 为了校验拟合的效果而生成的系列参数
    global ka
    global curveParam
    global ksRes
    global MSE
    global RMSE
    global R2
    ka = chi2_contingency(np.asarray([fitedData, scoreListN]))
    ksRes = kstest(scoreListN, fitedData)
    MSE = mean_squared_error(fitedData, scoreListN)
    RMSE = np.sqrt(MSE)
    R2 = r2_score(fitedData, scoreListN)
    curveParam = popt
    return limit1
    # averageScore = average

def drawPic():
    # 绘图部分
    print()
    global scoreList
    xplot = [x for x in range(1, len(scoreList) + 1)]
    scoreList = sorted(scoreList)
    # plt.scatter(xplot, scoreList, s=2, color="r", marker=".", linewidth=1, alpha=0.3, label='疫情相关度')
    starList1 = [math.log10(x + 1) * 15 for x in starList]
    commentList1 = sorted([math.log10(x + 1) * 8 for x in commentList])
    forwardList1 = [math.log10(x + 1) * 10 for x in forwardList]
    # plt.plot(xplot, starList1, color="g", linestyle="", marker="^", linewidth=1, label='点赞数')
    plt.plot(xplot, commentList1, color="y", linestyle="", marker=".", linewidth=1, alpha=0.1, label='评论数')
    # plt.plot(xplot, forwardList1, color="y", linestyle="", marker="^", linewidth=1, label='转发数')
    plt.rcParams['figure.dpi'] = 300
    plt.xlabel = '微博数量（从左到右递增）'
    plt.title('阶段' + str(stageNo) + '疫情相关度与拟合相关度结果')
    plt.legend()
    plt.savefig(path + '/SaveTest.png', dpi=300)

    # 绘制热度和疫情相关度之间的联系
    # sortedWeiboList = sorted(allWeibo, key=lambda x:x[''])
    # plt.show()

# 正态分布的函数表达式
def gaussian(x, *param):
    return param[1] * np.exp(-np.power(x - param[0], 2.) / (2 * np.power(param[1], 2.)))

# 正态分布的函数表达式
def exponent(x, a, b, c, d):
    return a * np.power(b, (x - d)) + c

def keywordsInit():
    """
        前面的代码是用来生成在人工筛选之前的各个关键词的权值表的
    """
    with open(stage.stage[stageNo]['path'] +'/keywords-' + stage.stage[stageNo]['Path'] +'.json', mode='r', encoding='utf-8') as fp:
        keywords = json.load(fp)
        keywords = sorted(keywords, key=lambda s: s[1], reverse=True)
        maxNum = math.log10(keywords[0][1])
        for item in keywords:
            item[1] = math.log10(item[1])
            item[1] = item[1] / maxNum
        newWords = json.dumps(keywords, indent=4, separators=(',', ':'), ensure_ascii=False)
        with open(stage.stage[stageNo]['path'] +'/COVkeywords-' + stage.stage[stageNo]['Path'] +'.json', mode='w', encoding='utf-8') as fp2:
            fp2.write(newWords)
            fp2.flush()
    """
        下面这部分代码是在第一步坐完，人工筛选过一次关键词后的结果
    """
    # with open(stage.stage[stageNo]['path'] + '/COVkeywords-' + stage.stage[stageNo]['Path'] + '-.json', mode='r',
    #           encoding='utf-8') as fp2:
    #     # print(fp2.read())
    #     keywords = json.loads(str(fp2.read()))
    # return keywords

def getSingleBlog(fp):
    blog = ""
    tmp = fp.readline()
    if tmp[0] == '{':
        while tmp[0] != '}':
            blog = blog + tmp
            tmp = fp.readline()
        blog += '}'
    return blog

def splitWeibo():
    date = beginDate
    with open(path + '/stageCOVWeibo.json', mode='r', encoding='utf-8') as fp:
        allCovWeibo_ = json.load(fp)
        allCovWeibo = sorted(allCovWeibo_, key=lambda x: x['发布时间'])
        fp2 = open(path + '/' + date + '/' + date + 'blog-COV.json', 'w', encoding='utf-8')
        fp2.write('[\n')
        numOfDate = 0
        numOfCurDateCOVWeibo = 0
        numOfCOVWeibo = len(allCovWeibo)
        dailyCOVWeiboRatioList = []
        curDateCOVWeiboList = []
        for i in range(0, numOfCOVWeibo):
            # print(i)
            if (allCovWeibo[i]['发布时间'] + '-') == date:
                # weiboS = json.dumps(allCovWeibo[i], indent=4, separators=(',', ':'), ensure_ascii=False)
                # fp2.write(weiboS)
                # if i != (numOfCOVWeibo - 1) and ((i + 1) <= (numOfCOVWeibo - 1) and (allCovWeibo[i + 1]['发布时间']+'-') == date):
                #     # print('next ' + allCovWeibo[i + 1]['发布时间']+'-')
                #     fp2.write(',\n')
                # else:
                #     # print('NO,')
                #     fp2.write('\n')
                curDateCOVWeiboList.append(allCovWeibo[i])
                numOfCurDateCOVWeibo += 1
            else:
                # fp2.write(']')
                fp2.write(json.dumps(curDateCOVWeiboList, indent=4, separators=(',', ':'), ensure_ascii=False))

                fp3 = open(path + '/' + date + '/' + date + 'blogInfo.json', mode='r', encoding='utf-8')
                curDateWeiboInfo = json.load(fp3)
                curDateWeiboInfo['疫情相关微博数量'] = numOfCurDateCOVWeibo
                # curDateWeiboInfo['疫情信息热度'] = 0
                curDateWeiboInfo['疫情相关微博占比'] = numOfCurDateCOVWeibo / curDateWeiboInfo['微博数量']
                dailyCOVWeiboRatioList.append(curDateWeiboInfo['疫情相关微博占比'])
                print(dailyCOVWeiboRatioList)
                COVWeiboNumList.append(numOfCurDateCOVWeibo)
                numOfCurDateCOVWeibo = 0
                outWeiboInfo = json.dumps(curDateWeiboInfo, indent=4, separators=(',', ':'), ensure_ascii=False)
                fp3 = open(path + '/' + date + '/' + date + 'blogInfo.json', mode='w', encoding='utf-8')
                fp3.write(outWeiboInfo)
                fp3.flush()
                print(date + ' Finished')
                # break

                date = allCovWeibo[i]['发布时间'] + '-'
                fp2 = open(path + '/' + date + '/' + date + 'blog-COV.json', 'w', encoding='utf-8')
                # fp2.write('[\n')
                curDateCOVWeiboList = []
                # weiboS = json.dumps(allCovWeibo[i], indent=4, separators=(',', ':'), ensure_ascii=False)
                curDateCOVWeiboList.append(allCovWeibo[i])
                # if i != (numOfCOVWeibo - 1) or ((i + 1) <= (numOfCOVWeibo - 1) and (allCovWeibo[i]['发布时间']+'-') == date):
                #     fp2.write(',\n')
                # else:
                #     # print('NO,')
                #     fp2.write('\n]')
                numOfDate += 1
                # if numOfDate > 3:
                #     break
        fp3 = open(path + '/' + date + '/' + date + 'blogInfo.json', mode='r', encoding='utf-8')
        curDateWeiboInfo = json.load(fp3)
        curDateWeiboInfo['疫情相关微博数量'] = numOfCurDateCOVWeibo
        curDateWeiboInfo['疫情相关微博占比'] = numOfCurDateCOVWeibo / curDateWeiboInfo['微博数量']
        COVWeiboNumList.append(numOfCurDateCOVWeibo)
        dailyCOVWeiboRatioList.append(curDateWeiboInfo['疫情相关微博占比'])
        outWeiboInfo = json.dumps(curDateWeiboInfo, indent=4, separators=(',', ':'), ensure_ascii=False)
        fp3 = open(path + '/' + date + '/' + date + 'blogInfo.json', mode='w', encoding='utf-8')
        fp3.write(outWeiboInfo)
        fp3.flush()
        print(date + ' Finished')

        # 绘图
        plt.clf()
        llen = len(dailyCOVWeiboRatioList) + 1
        xplot = [x for x in range(1, llen)]
        dailyCOVWeiboRatioList_ = [x * 100 for x in dailyCOVWeiboRatioList]
        plt.xticks(np.arange(0, llen + 2, 2))
        plt.plot(xplot, dailyCOVWeiboRatioList_, color="r", marker=".", linewidth=1, alpha=0.7, label='疫情相关微博百分比')
        for a, b in zip(xplot, dailyCOVWeiboRatioList_):
            plt.text(a, b, (a, round(b, 2)), ha='center', va='bottom', fontsize=10)
        # global COVWeiboNumList
        # print('ANALYSE')
        # print(xplot)
        # print(COVWeiboNumList)
        plt.plot(xplot, COVWeiboNumList, color="g", marker=".", linewidth=1, alpha=0.7, label='疫情相关微博数量')
        plt.title('阶段' + str(stageNo) + '每日疫情相关微博数量与当日占比')
        plt.legend()
        plt.savefig(path + '/ratioByDate.png', dpi=300)

def verifyJSON():
    date = beginDate
    while Date.dateCmp(date, endDate) != 1:
        try:
            fp2 = open(path + '/' + date + '/' + date + 'blog-COV.json', 'r', encoding='utf-8')
            json.load(fp2)
            print(date + ' Finished!')
        except Exception as e:
            print(date)
            print(e)
        date = Date.getNextDate(date)

if __name__ == "__main__":
    keywordsInit()
    # print(exponent(2, 5, 6))
    # run()
    # splitWeibo()
    # verifyJSON()

max_issues_repo_name: Ironstarboy/DataSciBasic

max_forks_repo_name: Ironstarboy/DataSciBasic


text:
import h3
import numpy as np
import torch

from graph_weather import GraphWeatherAssimilator, GraphWeatherForecaster
from graph_weather.models import AssimilatorDecoder, AssimilatorEncoder, Decoder, Encoder, Processor
from graph_weather.models.losses import NormalizedMSELoss

def test_encoder():
    lat_lons = []
    for lat in range(-90, 90, 5):
        for lon in range(0, 360, 5):
            lat_lons.append((lat, lon))
    model = Encoder(lat_lons).eval()

    features = torch.randn((2, len(lat_lons), 78))
    with torch.no_grad():
        x, edge_idx, edge_attr = model(features)
    assert x.size() == (5882 * 2, 256)
    assert edge_idx.size() == (2, 41162 * 2)

def test_encoder_uneven_grid():
    lat_lons = []
    for lat in range(-90, 90, 7):
        for lon in range(0, 180, 5):
            lat_lons.append((lat, lon))
        for lon in range(180, 360, 9):
            lat_lons.append((lat, lon))
    model = Encoder(lat_lons).eval()

    features = torch.randn((2, len(lat_lons), 78))
    with torch.no_grad():
        x, edge_idx, edge_attr = model(features)
    assert x.size() == (5882 * 2, 256)
    assert edge_idx.size() == (2, 41162 * 2)

def test_assimilation_encoder_uneven_grid():
    lat_lons = []
    for lat in range(-90, 90, 7):
        for lon in range(0, 180, 5):
            lat_lons.append((lat, lon, np.random.random(1)))
        for lon in range(180, 360, 9):
            lat_lons.append((lat, lon, np.random.random(1)))
    model = AssimilatorEncoder().eval()

    features = torch.randn((2, len(lat_lons), 2))
    with torch.no_grad():
        x, edge_idx, edge_attr = model(features, torch.tensor(lat_lons))
    assert x.size() == (5882 * 2, 256)
    assert edge_idx.size() == (2, 41162 * 2)

def test_processor():
    processor = Processor().eval()
    lat_lons = []
    for lat in range(-90, 90, 5):
        for lon in range(0, 360, 5):
            lat_lons.append((lat, lon))
    model = Encoder(lat_lons).eval()

    features = torch.randn((3, len(lat_lons), 78))
    with torch.no_grad():
        x, edge_idx, edge_attr = model(features)
        out = processor(x, edge_idx, edge_attr)
    assert out.size() == x.size()

def test_decoder():
    lat_lons = []
    for lat in range(-90, 90, 5):
        for lon in range(0, 360, 5):
            lat_lons.append((lat, lon))
    model = Decoder(lat_lons).eval()
    features = torch.randn((3, len(lat_lons), 78))
    processed = torch.randn((3 * h3.num_hexagons(2), 256))
    with torch.no_grad():
        x = model(processed, features.shape[0])
    assert x.size() == (3, 2592, 78)

def test_assimilator():
    lat_lons = []
    for lat in range(-90, 90, 5):
        for lon in range(0, 360, 5):
            lat_lons.append((lat, lon))
    model = AssimilatorDecoder(lat_lons).eval()
    processed = torch.randn((3 * h3.num_hexagons(2), 256))
    with torch.no_grad():
        x = model(processed, 3)
    assert x.size() == (3, 2592, 78)

def test_end2end():
    lat_lons = []
    for lat in range(-90, 90, 5):
        for lon in range(0, 360, 5):
            lat_lons.append((lat, lon))
    model = Encoder(lat_lons).eval()
    processor = Processor().eval()
    decoder = Decoder(lat_lons).eval()
    features = torch.randn((4, len(lat_lons), 78))
    with torch.no_grad():
        x, edge_idx, edge_attr = model(features)
        out = processor(x, edge_idx, edge_attr)
        pred = decoder(out, features.shape[0])
    assert pred.size() == (4, 2592, 78)

def test_forecaster():
    lat_lons = []
    for lat in range(-90, 90, 5):
        for lon in range(0, 360, 5):
            lat_lons.append((lat, lon))
    model = GraphWeatherForecaster(lat_lons)

    features = torch.randn((1, len(lat_lons), 78))

    out = model(features)
    assert not torch.isnan(out).any()
    assert not torch.isnan(out).any()

def test_assimilator_model():
    obs_lat_lons = []
    for lat in range(-90, 90, 7):
        for lon in range(0, 180, 6):
            obs_lat_lons.append((lat, lon, np.random.random(1)))
        for lon in 360 * np.random.random(100):
            obs_lat_lons.append((lat, lon, np.random.random(1)))

    output_lat_lons = []
    for lat in range(-90, 90, 5):
        for lon in range(0, 360, 5):
            output_lat_lons.append((lat, lon))
    model = GraphWeatherAssimilator(output_lat_lons=output_lat_lons, analysis_dim=24)

    features = torch.randn((1, len(obs_lat_lons), 2))
    lat_lon_heights = torch.tensor(obs_lat_lons)
    out = model(features, lat_lon_heights)
    assert not torch.isnan(out).any()
    assert not torch.isnan(out).any()
    loss.backward()

def test_forecaster_and_loss():
    lat_lons = []
    for lat in range(-90, 90, 5):
        for lon in range(0, 360, 5):
            lat_lons.append((lat, lon))
    criterion = NormalizedMSELoss(lat_lons=lat_lons, feature_variance=torch.randn((78,)))
    model = GraphWeatherForecaster(lat_lons)

    features = torch.randn((2, len(lat_lons), 78))

    out = model(features)
    assert not torch.isnan(out).any()
    assert not torch.isnan(out).any()


text:
import numpy as np
import pandas as pd
import torch
from get_symbol import get_symbol
from prediction_pipeline import *
from preprocessing import preprocess

def load_data(expression_path, libraries, gr_truth_path):
    norm = get_expression_data(expression_path)
    binary_matrix = get_binary_matrix(norm.index, libraries)
    gr_truth = get_gr_truth(norm.columns, gr_truth_path)
    list_inputs = prepare_data(norm, binary_matrix)
    return list_inputs, gr_truth

def get_expression_data(expression_path):
    """
    Get expression data from the file path, turn Ensembl number
    indexing into gene symbols, normalize data.
    """
    f = pd.read_csv(expression_path)
    f.index = f.iloc[:, 0] # Make ENSG genes as row indexing
    f = f.iloc[:, 1:] # Remove first index column
    # Normalize data
    norm = preprocess(f)
    # Convert Ensembl number index to gene symbol
    norm = get_symbol(norm)
    return norm

def get_binary_matrix(gene_expr, libraries):
    """
    Get binary matrix with genes as rows and pathways as columns.
    If a gene is found in a given pathway, it is given a value of
    1. Else, 0. Only the list of genes in common between that found
    in the gene set libraries and the current dataset are used.
    """
    function_to_genes = {}
    set_genes = set()
    for lib in libraries:
        f2g, genes = gene_set_dictionaries(lib)
        function_to_genes.update(f2g)
        set_genes = set_genes | set(genes)
    common_genes = list(set_genes & set(gene_expr))
    binary_matrix = gs_binary_matrix(function_to_genes, set_genes).loc[common_genes]
    return binary_matrix

def get_gr_truth(list_samples, gr_truth_path):
    gr_truth = pd.read_csv(gr_truth_path, sep='\t')
    gr_truth = gr_truth.iloc[10, 1:]
    idx = [ "_".join(i.split("_")[:2]) for i in gr_truth.index ]
    gr_truth.index = idx
    gr_truth = pd.DataFrame(gr_truth).loc[list_samples]
    test = [ 1.0 if res[-3:] == 'POS' else 0.0 for res in gr_truth.iloc[:, 0] ]
    gr_truth.iloc[:, 0] = test
    gr_truth.columns = ["Truth"]
    return gr_truth

def prepare_data(norm, binary_matrix):
    """
    For each column of the expression matrix, multiply the
    smaller binary matrix by the common genes to get a matrix that is
    basically the same as the binary matrix but with expression instead of
    1's. This leads to a list of matrices.
    """
    list_inputs = []
    bm_tens = torch.tensor(binary_matrix.T.values)
    for sample in norm.columns:
        item = torch.mul(bm_tens, torch.tensor(norm.loc[common_genes, sample].values))
        list_inputs.append(np.array(item).astype(np.float32))
    return list_inputs