text:
import numpy as np
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt

from collections import Counter
from collections import defaultdict

from attr import attributes
from langdetect import detect
from scipy.spatial.distance import pdist, squareform
from sklearn.cluster import DBSCAN
from sklearn.model_selection import train_test_split
from sklearn.neighbors import NearestNeighbors
from sklearn.preprocessing import MinMaxScaler
from statsmodels.compat import pandas

import utils
from pathlib import Path

df = utils.load("../data/tracks.csv", dummies=True, buckets="basic", fill=True)

df.info()

column2drop = [
    ("album", "tags"),
    ("artist", "tags"),
    ("track", "tags"),
    ("track", "genres"),
    ("track", "genres_all"),
    ("track", "license"),
    ("track", "language_code"),
    ("album", "title"),
    ("artist", "name"),
    ("set", "split"),
    ("track", "title"),
    ("album", "id"),
    ("artist", "id"),
    ("album", "date_created"),
]
df.drop(column2drop, axis=1, inplace=True)

# Riformatto le date

df["artist", "date_created"] = df["artist", "date_created"].astype("Int64")
# stampo qualche valore
df.info()
print(df["artist", "date_created"].unique())
print(df["album", "listens"].unique())
print(df["album", "information"].unique())

class_name = ("album", "type")
attributes = [col for col in df.columns if col != class_name]
X = df[attributes].values
y = df[class_name]

"""BOXPLOT PER OGNUNA DELLE 26 VARIABILI CHE VOGLIO TENERE
df[("album", "comments")].plot.box()
plt.xlabel(("album", "comments"))
plt.show()
df[("album", "favorites")].plot.box()
plt.xlabel(("album", "favorites"))
plt.show()
df[("album", "listens")].plot.box()
plt.xlabel(("album", "listens"))
plt.show()
df[("album", "tracks")].plot.box()
plt.xlabel(("album", "tracks"))
plt.show()
df[("album", "engineer")].plot.box() #da levare
plt.xlabel(("album", "engineer"))
plt.show()
df[("album", "information")].plot.box() #da levare
plt.xlabel(("album", "information"))
plt.show()
df[("album", "producer")].plot.box() #da levare
plt.xlabel(("album", "producer"))
plt.show()
df[("album", "type")].plot.box()
plt.xlabel(("album", "type"))
plt.show()

df[("artist", "comments")].plot.box()
plt.xlabel(("artist", "comments"))
plt.show()
df[("artist", "date_created")].plot.box()
plt.xlabel(("artist", "date_created"))
plt.show()
df[("artist", "favorites")].plot.box()
plt.xlabel(("artist", "favorites"))
plt.show()
df[("artist", "active_year_end")].plot.box() #da levare
plt.xlabel(("artist", "active_year_end"))
plt.show()
df[("artist", "wikipedia_page")].plot.box() #da levare
plt.xlabel(("artist", "wikipedia_page"))
plt.show()
df[("artist", "bio")].plot.box() #da levare
plt.xlabel(("artist", "bio"))
plt.show()
df[("artist", "website")].plot.box() #dalevare
plt.xlabel(("artist", "website"))
plt.show()

df[("track", "comments")].plot.box()
plt.xlabel(("track", "comments"))
plt.show()
df[("track", "date_created")].plot.box()
plt.xlabel(("track", "date_created"))
plt.show()
df[("track", "duration")].plot.box()
plt.xlabel(("track", "duration"))
plt.show()
df[("track", "favorites")].plot.box()
plt.xlabel(("track", "favorites"))
plt.show()
df[("track", "interest")].plot.box()
plt.xlabel(("track", "interest"))
plt.show()
df[("track", "listens")].plot.box()
plt.xlabel(("track", "listens"))
plt.show()
df[("track", "number")].plot.box()
plt.xlabel(("track", "number"))
plt.show()
df[("track", "composer")].plot.box() #dalevare
plt.xlabel(("track", "composer"))
plt.show()
df[("track", "date_recorded")].plot.box()#dalevare
plt.xlabel(("track", "date_recorded"))
plt.show()
df[("track", "information")].plot.box()#da levare
plt.xlabel(("track", "information"))
plt.show()
df[("track", "lyricist")].plot.box() #da levare
plt.xlabel(("track", "lyricist"))
plt.show()
df[("track", "publisher")].plot.box() #da levare
plt.xlabel(("track", "publisher"))
plt.show()
"""

# elimino le colonne che ho selezionato
column2drop = [
    ("album", "engineer"),
    ("album", "information"),
    ("album", "producer"),
    ("artist", "active_year_end"),
    ("artist", "wikipedia_page"),
    ("artist", "website"),
    ("artist", "bio"),
    ("track", "composer"),
    ("track", "date_recorded"),
    ("track", "information"),
    ("track", "lyricist"),
    ("track", "publisher"),
    ("artist", "date_created"),
    ("track", "date_recorded"),
    ("track", "date_created"),
]
df.drop(column2drop, axis=1, inplace=True)
# normalizzo in basic
normalized_df = (df - df.min()) / (df.max() - df.min())
df = normalized_df
# FACCIO  IL PLOTTING BOXPLOT del Df completo
plt.figure(figsize=(20, 25))
b = sns.boxplot(data=df, orient="h")
b.set(ylabel="Class", xlabel="Normalization Value")
plt.show()
"""
# PLOT per trovare best esp
sns.set()
neigh = NearestNeighbors(n_neighbors=3)
nbrs = neigh.fit(X)
distances, indices = nbrs.kneighbors(X)
distances = np.sort(distances, axis=0)
distances = distances[:, 1]
plt.plot(distances)
plt.show()
"""
"""APPLICO IL DBSCAN """

from sklearn.cluster import DBSCAN

print("DBSCAN")
dbscan = DBSCAN(eps=0.00000001, min_samples=2)
dbscan = dbscan.fit(X)
labels = dbscan.labels_
# Number of clusters in labels, ignoring noise if present.
n_clusters_ = len(set(labels)) - (1 if -1 in labels else 0)
n_noise_ = list(labels).count(-1)

print("Estimated number of clusters: %d" % n_clusters_)
print("Estimated number of noise points: %d" % n_noise_)

print(df.loc[(labels == -1)])
"""
# Calcolo eps e min samples migliori
eps_to_test = [3, 4, 5, 6, 7, 8]  # i migliori sono esp 4 min 3
min_samples_to_test = [2, 3, 4, 5]

print("EPS:", list(eps_to_test))
print("MIN_SAMPLES:", list(min_samples_to_test))

def get_metrics(eps, min_samples, dataset, iter_):
    # Fitting ======================================================================

    dbscan = DBSCAN(eps=eps, min_samples=min_samples)
    dbscan.fit(dataset)

    # Mean Noise Point Distance metric =============================================
    noise_indices = dbscan.labels_ == -1

    if True in noise_indices:
        neighboors = NearestNeighbors(n_neighbors=6).fit(dataset)
        distances, indices = neighboors.kneighbors(dataset)
        noise_distances = distances[noise_indices, 1:]
        noise_mean_distance = round(noise_distances.mean(), 3)
    else:
        noise_mean_distance = None

    # Number of found Clusters metric ==============================================

    number_of_clusters = len(set(dbscan.labels_[dbscan.labels_ >= 0]))

    # Log ==========================================================================

    print(
        "%3d | Tested with eps = %3s and min_samples = %3s | %5s %4s"
        % (iter_, eps, min_samples, str(noise_mean_distance), number_of_clusters)
    )

    return (noise_mean_distance, number_of_clusters)

# Dataframe per la metrica sulla distanza media dei noise points dai K punti più vicini
results_noise = pd.DataFrame(
    data=np.zeros((len(eps_to_test), len(min_samples_to_test))),  # Empty dataframe
    columns=min_samples_to_test,
    index=eps_to_test,
)

# Dataframe per la metrica sul numero di cluster
results_clusters = pd.DataFrame(
    data=np.zeros((len(eps_to_test), len(min_samples_to_test))),  # Empty dataframe
    columns=min_samples_to_test,
    index=eps_to_test,
)

iter_ = 0

print("ITER| INFO%s |  DIST    CLUS" % (" " * 39))
print("-" * 65)

for eps in eps_to_test:
    for min_samples in min_samples_to_test:
        iter_ += 1

        # Calcolo le metriche
        noise_metric, cluster_metric = get_metrics(eps, min_samples, df, iter_)

        # Inserisco i risultati nei relativi dataframe
        results_noise.loc[eps, min_samples] = noise_metric
        results_clusters.loc[eps, min_samples] = cluster_metric

fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(16, 8))

sns.heatmap(results_noise, annot=True, ax=ax1, cbar=False).set_title(
    "METRIC: Mean Noise Points Distance"
)
sns.heatmap(results_clusters, annot=True, ax=ax2, cbar=False).set_title(
    "METRIC: Number of clusters"
)

ax1.set_xlabel("N")
ax2.set_xlabel("N")
ax1.set_ylabel("EPSILON")
ax2.set_ylabel("EPSILON")

plt.tight_layout()
plt.show()
"""

max_issues_repo_name: Mariabba/fma-datamining

max_forks_repo_name: Mariabba/fma-datamining

reason: import numpy,from scipy,from statsmodels


text:
#   Copyright (c) 2021 PaddlePaddle Authors. 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.
import numpy as np
from op_test import OpTest
import paddle.fluid as fluid
from paddle.fluid import core
import unittest
import paddle
paddle.enable_static()

class Decoder(object):
    def __init__(self, transitions, use_tag=True):
        self.transitions = transitions
        self.use_tag = use_tag
        self.start_idx, self.stop_idx = -1, -2

    def __call__(self, inputs, length):
        bs, seq_len, n_label = inputs.shape
        inputs_t = np.transpose(inputs, (1, 0, 2))
        trans_exp = np.expand_dims(self.transitions, axis=0)
        historys = []
        left_length = np.array(length)
        max_seq_len = np.amax(left_length)
        left_length = np.expand_dims(left_length, 1)
        alpha = np.full((bs, n_label), -1e4, dtype='float32') if self.use_tag \
            else np.zeros((bs, n_label), dtype='float32')
        alpha[:, -1] = 0
        for i, logit in enumerate(inputs_t[:max_seq_len]):
            if i == 0 and not self.use_tag:
                alpha = logit
                left_length = left_length - 1
                continue
            alpha_exp = np.expand_dims(alpha, 2)
            alpha_trn_sum = alpha_exp + trans_exp
            max_res = np.amax(alpha_trn_sum, 1), np.argmax(alpha_trn_sum, 1)
            historys = historys + [max_res[1]] if i >= 1 else []
            alpha_nxt = max_res[0] + logit
            mask = (left_length > 0)
            alpha = mask * alpha_nxt + (1 - mask) * alpha
            if self.use_tag:
                alpha += (left_length == 1) * trans_exp[:, self.stop_idx]
            left_length = left_length - 1
        scores, last_ids = np.amax(alpha, 1), np.argmax(alpha, 1)
        left_length = left_length[:, 0]
        last_ids_update = last_ids * (left_length >= 0)
        batch_path = [last_ids_update]
        batch_offset = np.arange(bs) * n_label
        for hist in reversed(historys):
            left_length = left_length + 1
            gather_idx = batch_offset + last_ids
            last_ids_update = np.take(hist, gather_idx) * (left_length > 0)
            mask = (left_length == 0)
            last_ids_update = last_ids_update * (1 - mask) + last_ids * mask
            batch_path.insert(0, last_ids_update)
            last_ids = last_ids_update + (left_length < 0) * last_ids
        batch_path = np.stack(batch_path, 1)
        return scores, batch_path

class TestViterbiOp(OpTest):
    def set_attr(self):
        self.dtype = "float32" if core.is_compiled_with_rocm() else "float64"
        self.use_tag = True
        self.bz, self.len, self.ntags = 4, 8, 10

    def setUp(self):
        self.op_type = "viterbi_decode"
        self.python_api = paddle.text.viterbi_decode
        self.set_attr()
        bz, length, ntags = self.bz, self.len, self.ntags
        self.input = np.random.randn(bz, length, ntags).astype(self.dtype)
        self.trans = np.random.randn(ntags, ntags).astype(self.dtype)
        self.length = np.random.randint(1, length + 1, [bz]).astype('int64')
        decoder = Decoder(self.trans, self.use_tag)
        scores, path = decoder(self.input, self.length)
        self.inputs = {
            'Input': self.input,
            'Transition': self.trans,
            'Length': self.length
        }
        self.attrs = {'include_bos_eos_tag': self.use_tag, }
        self.outputs = {'Scores': scores, 'Path': path}

    def test_output(self):
        self.check_output(check_eager=True)

class TestViterbiAPI(unittest.TestCase):
    def set_attr(self):
        self.use_tag = True
        self.bz, self.len, self.ntags = 4, 8, 10
        self.places = [fluid.CPUPlace(), fluid.CUDAPlace(0)]    \
            if core.is_compiled_with_cuda() else [fluid.CPUPlace()]

    def setUp(self):
        self.set_attr()
        bz, length, ntags = self.bz, self.len, self.ntags
        self.input = np.random.randn(bz, length, ntags).astype('float32')
        self.transitions = np.random.randn(ntags, ntags).astype('float32')
        self.length = np.random.randint(1, length + 1, [bz]).astype('int64')
        decoder = Decoder(self.transitions, self.use_tag)
        self.scores, self.path = decoder(self.input, self.length)

    def check_static_result(self, place):
        bz, length, ntags = self.bz, self.len, self.ntags
        with fluid.program_guard(fluid.Program(), fluid.Program()):
            Input = fluid.data(
                name="Input", shape=[bz, length, ntags], dtype="float32")
            Transition = fluid.data(
                name="Transition", shape=[ntags, ntags], dtype="float32")
            Length = fluid.data(name="Length", shape=[bz], dtype="int64")
            decoder = paddle.text.ViterbiDecoder(Transition, self.use_tag)
            score, path = decoder(Input, Length)
            exe = fluid.Executor(place)
            feed_list = {
                "Input": self.input,
                "Transition": self.transitions,
                "Length": self.length
            }
            fetches = exe.run(feed=feed_list, fetch_list=[score, path])
            np.testing.assert_allclose(fetches[0], self.scores, rtol=1e-5)
            np.testing.assert_allclose(fetches[1], self.path)

    def test_static_net(self):
        for place in self.places:
            self.check_static_result(place)

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

ext: py

max_stars_repo_name: RangeKing/Paddle

max_issues_repo_name: RangeKing/Paddle

max_forks_repo_name: RangeKing/Paddle

reason: import numpy


text:
#!/usr/bin/env python
# coding: utf-8
##############################################################################
##                 How well can you control your thoughts?                  ##
##             The ability to modulate the DMN told in two parts:           ##
##                        Feedback and No-Feedback                          ##
##                                                                          ##
##                        R. Cameron Craddock, PhD                          ##
##                        Jonathan M. Lisinski, MS                          ##
##                         Stephen M. LaConte, PhD                          ##
##############################################################################

############################
#  Import various modules  #
############################

import VisionEgg
VisionEgg.start_default_logging(); VisionEgg.watch_exceptions()

from VisionEgg.Core import *
from VisionEgg.FlowControl import Presentation, Controller, FunctionController
from VisionEgg.MoreStimuli import *
from VisionEgg.Textures import *
from math import *
import pygame
import OpenGL.GL as gl
from VisionEgg.DaqKeyboard import *
from VisionEgg.Text import *
from VisionEgg.WrappedText import *
from VisionEgg.ResponseControl import *
from string import *
import Image, ImageDraw # Python Imaging Library (PIL)
import os, sys

# to interact with the Lumina box
import pyxid

try:
    import Tkinter
except:
    pass

############################
# Define                   #
############################

LUMINA        = 0
DETREND       = 0
TCPIP         = 1    # Set 1 to receive data over TCP/IP
TCPIP_PORT    = 8000 # Port receiving TCP/IP data
TCPIP_GUI     = 1    # Open GUI to initialize TCP/IP

#AFNI_IP='localhost'
#AFNI_PORT=5000
#AFNI_COM_GUI=1

ALPHA_MAX = 0.5
ALPHA_SCALE = 1.0

LUMINA_TRIGGER=4

# this is nparadigms x 2, where the second dimension is no feedback, and
# feedback (no feedback = 0, feedback = 1)
stim_files= \
    [ \
     ['DMN_TRACKING_nofb_normal_wander_v2.cfg',\
      'DMN_TRACKING_fb_normal_wander_v2.cfg'],\
     ['DMN_TRACKING_nofb_normal_focus_v2.cfg', \
      'DMN_TRACKING_fb_normal_focus_v2.cfg'], \
     ['DMN_TRACKING_nofb_swapped_wander_v2.cfg', \
      'DMN_TRACKING_fb_swapped_wander_v2.cfg'], \
     ['DMN_TRACKING_nofb_swapped_focus_v2.cfg', \
     'DMN_TRACKING_fb_swapped_focus_v2.cfg'], \
     ['SIMULATION_DMN_TRACKING_nofb_normal_wander_v2.cfg',\
      'SIMULATION_DMN_TRACKING_fb_normal_wander_v2.cfg'],\
     ['SIMULATION_DMN_TRACKING_nofb_normal_focus_v2.cfg', \
      'SIMULATION_DMN_TRACKING_fb_normal_focus_v2.cfg'], \
     ['SIMULATION_DMN_TRACKING_nofb_swapped_wander_v2.cfg', \
      'SIMULATION_DMN_TRACKING_fb_swapped_wander_v2.cfg'], \
     ['SIMULATION_DMN_TRACKING_nofb_swapped_focus_v2.cfg', \
     'SIMULATION_DMN_TRACKING_fb_swapped_focus_v2.cfg'], \
    ]

# default server address
server_address=('127.0.0.1','%d'%(TCPIP_PORT))

####

if DETREND:
    from scipy import polyfit # for detrending

# parse out the name of this program and the paradigm
# name
prog_name = os.path.basename( sys.argv[0] )
pdigm_name = split( os.path.basename( sys.argv[0]), '.')[0]

###############################
# TK window to get parameters #
# copied from the code in     #
# VisionEgg.TCPSERVER         #
###############################
class GetPdigmConfigWindow(Tkinter.Frame):
    def __init__(self,server_address,**kw):
        try:
            Tkinter.Frame.__init__(self,**kw)
        except AttributeError,x:
            # restart Tk and see if that helps
            tk=Tkinter.Tk()
            Tkinter.Frame.__init__(self,tk,**kw)
        self.winfo_toplevel().title("%s Configuration"%(pdigm_name))
        self.server_address = server_address
        hostname,port = self.server_address
        self.clicked_ok = 0
        self.subjid  = Tkinter.StringVar()
        self.subjid.set('')
        self.fb  = Tkinter.StringVar()
        self.fb.set("-1")
        self.pdigm  = Tkinter.StringVar()
        self.pdigm.set("-1")
        self.hostname = Tkinter.StringVar()
        self.hostname.set(hostname)
        self.port = Tkinter.StringVar()
        self.port.set(port)
        row = 0
        Tkinter.Label(self,
            text="Subject ID:",\
            ).grid(row=row,column=0,sticky=Tkinter.E)
        Tkinter.Entry(self,textvariable=self.subjid).grid(row=row,column=1,\
            sticky=Tkinter.W+Tkinter.E,padx=10)
        row += 1
        Tkinter.Label(self,
            text="Feedback 0 (off) or 1 (on):",\
            ).grid(row=row,column=0,sticky=Tkinter.E)
        Tkinter.Entry(self,textvariable=self.fb).grid(row=row,column=1,\
            sticky=Tkinter.W+Tkinter.E,padx=10)
        row += 1
        Tkinter.Label(self,
            text="Paradigm number (0,1,2,4):",\
            ).grid(row=row,column=0,sticky=Tkinter.E)
        Tkinter.Entry(self,textvariable=self.pdigm).grid(row=row,column=1,\
            sticky=Tkinter.W+Tkinter.E,padx=10)

        if TCPIP:
            row += 1
            Tkinter.Label(self,
                text="Please enter the hostname and port you would like to listen"+\
                     " for connections on.",).grid(row=row,columnspan=2)
            row += 1
            Tkinter.Label(self,
                text="Hostname (blank means localhost):",\
                ).grid(row=row,column=0,sticky=Tkinter.E)
            Tkinter.Entry(self,textvariable=self.hostname).grid(row=row,column=1,\
                sticky=Tkinter.W+Tkinter.E,padx=10)
            row += 1
            Tkinter.Label(self, text="Port:",).grid(row=row,column=0,sticky=Tkinter.E)
            Tkinter.Entry(self,textvariable=self.port).grid(row=row,column=1,\
                sticky=Tkinter.W+Tkinter.E,padx=10)
            row += 1
            b = Tkinter.Button(self,text="Bind port and listen for connections",\
                command=self.click_ok)
        else:
            row += 1
            b = Tkinter.Button(self,text="OK",\
                command=self.click_ok)

        b.grid(row=row,columnspan=2)
        b.focus_force()
        b.bind('<Return>',self.click_ok)

    def click_ok(self,dummy_arg=None):
        subjid=self.subjid.get()

        try:
            fb=int(self.fb.get())
            print "setting fb as an int?? %d"%(fb)
        except:
            print "setting fb as a string?? %s"%(fb)
            fb=self.fb.get()

        try:
            pdigm=int(self.pdigm.get())
        except:
            pdigm=self.pdigm.get()

        if TCPIP:
            hostname = self.hostname.get()
            try:
                port = int(self.port.get())
            except:
                port = self.port.get()
            self.server_address = (hostname,port)
        self.clicked_ok = 1
        self.winfo_toplevel().destroy()

#bound = 0
#if not bound:
#  while not bound: # don't loop until the code is cleaner
    #if confirm_address_with_gui and self.dialog_ok:
         #window = GetServerAddressWindow(server_address)
         #window.pack()
         #window.mainloop()
         #if not window.clicked_ok:
             #return # User wants to quit
         #server_address = window.server_address
     #self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     #self.server_socket.bind(server_address)
     #bound = 1

# check to make sure that the required command line
# arguements were specified
if len(sys.argv) > 3:
    subjid = sys.argv[1]
    fb = int(sys.argv[2])
    pdigm = int(sys.argv[3])
    print "Executing: %s Subject ID: %s Feedback: " \
        "%d Paradigm: %d"%(prog_name, subjid, \
        fb, pdigm)
else:
    #print "Usage: %s <subjid> <feedback (0|1)> " \
        #"<pdigm index (1,2)>"%(prog_name)
    #sys.exit()
    window = GetPdigmConfigWindow(server_address)
    window.pack()
    window.mainloop()
    if not window.clicked_ok:
        sys.exit()

    subjid = window.subjid.get()

    try:
        fb = int(window.fb.get())
    except Exception as e:
        print "Error, feedback must be either a '1' or '0' " + e.message
        sys.exit(0)

    try:
        pdigm = int(window.pdigm.get())
    except:
        print "Error, paradigm must be an integer"
        sys.exit(0)

    if TCPIP:
        server_address = window.server_address
    print "Executing: %s Subject ID: %s Feedback: " \
        "%d Paradigm: %d"%(prog_name, subjid, \
        fb, pdigm)

if not subjid:
    print "invalid subject ID: %s"%(subjid)
    sys.exit()

# check the sanity of the command line arguements
if fb < 0 or fb > 1:
    print "Feedback: %d should be either 0 (no feedback) " \
        "or 1 (feedback)"%(fb)
    sys.exit()

stim_path=os.path.dirname(os.path.abspath(sys.argv[0]))
# read in the stimulus labels
if pdigm < np.shape(stim_files)[0] and \
   fb < np.shape(stim_files)[1]:
    pfile=os.path.join(stim_path,stim_files[pdigm][fb])
else:
    print "Do not have an entry for paradigm %d (%d)"%(pdigm,fb)
    sys.exit()

# open the paradigm file
try:
    f=open(pfile,'r')
except Exception as e:
    print 'Could not open %s. %s'%( \
         pfile, e.message )
    sys.exit(1)
#except IOError as (errno, strerror):
    #print 'Could not open %s. I/O error({0}): {1}'%( \
         ##pfile,errno, strerror)
    #sys.exit(1)

# read in the files and parse
# out the meaning
Num_Stim = -666
ISI = -666.0
FEEDBACK = -666    # set to 0 to disable feedback
Instructions = ''
left_strings=[]
right_strings=[]
stimulus_strings=[]
show_array=[]

line=''
line_count=1
stim_count=0
try:
    for line in f:
        line=line.strip()
        if line and not '#' in line:
            if 'ISI' in line:
                ISI=float(line.split(':')[1])
            elif 'NUMSTIM' in line:
                Num_Stim=int(line.split(':')[1])
            elif 'FEEDBACK' in line:
                FEEDBACK=int(line.split(':')[1])
            elif 'INSTRUCTION' in line:
                if Instructions:
                    Instructions = Instructions + '\n' + \
                        line.split(':')[1]
                else:
                    Instructions = line.split(':')[1]
            else:
                # assume it is a stimulus specifier
                vals=line.split(';')
                #print vals
                if len(vals) > 3:
                    left_strings.append( vals[0] )
                    right_strings.append( vals[1] )
                    stimulus_strings.append( vals[2] )
                    show_array.append(int( vals[3] ))
                    stim_count=stim_count+1
                else:
                    print 'Improperly formatted STIM string %s (%d): %s (%d)' \
                        %( line, len(vals), pfile, line_count )
        line_count=line_count+1

except IOError:
    print '%s I/O error(%d)'%( \
        pfile)
    sys.exit(1)
#except IOError as (errno):
    #print '%s I/O error(%d)'%( \
        #pfile, errno)
    #sys.exit(1)

except ValueError:
    print '%s (%d) %s. Could not convert data to an integer.'%( \
       pfile, line_count, line )
    sys.exit(1)

except:
    print '%s (%d) %s. Unexpected error %s, maybe error in formatting?'%( \
       pfile, line_count, line, sys.exc_info()[0])
    sys.exit(1)

f.close()

if not Instructions:
    print "Error: INSTRUCTION not specified"
    sys.exit(1)

if Num_Stim == -666:
    print "Error: NUMSTIM not specified"
    sys.exit(1)

if Num_Stim < 0:
    print 'Error: Invalid NUMSTIM specified (%d)'%(Num_Stim)
    sys.exit(1)

if ISI == -666.0:
    print "Error: ISI not specified."
    sys.exit(1)

if ISI < 0:
    print 'Error: Invalid ISI specified (%g)'%(ISI)
    sys.exit(1)

if FEEDBACK != 0 and FEEDBACK != 1:
    print 'Error: Invalid FEEDBACK specified (%d)'%(FEEDBACK)
    sys.exit(1)

if stim_count < Num_Stim:
    print 'Error: Fewer stimulus specifiers (%d) than NUMSTIM (%d)'%( \
        stim_count,Num_Stim)
    sys.exit(1)

if len( left_strings ) < Num_Stim or len( right_strings ) < Num_Stim or \
   len( stimulus_strings ) < Num_Stim or len( show_array ) < Num_Stim:
    print 'Error: One of the stim lists (l %d, r %d, s %d, sh %d)' \
          ' is shorter than expected %d' %( \
        len(left_strings),len(right_strings),len(stimulus_strings), \
        len(show_array),Num_Stim)
    sys.exit(1)

Num_Secs = float(Num_Stim)*ISI
print 'Paradigm is %d * %g = %g seconds long'%(Num_Stim, ISI, Num_Secs)
print 'Instructions: %s'%(Instructions)
print 'Feedback: %d'%(FEEDBACK)

## create a log file
outpath=os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(sys.argv[0]))),
    'data',subjid)

if not os.path.isdir(outpath):
    os.makedirs(outpath)

filename = time.strftime ('%m-%d-%Y_%Hh-%Mm.txt');
filename = '_'.join ([subjid, 'NFB_FEEDBACK', filename])

try:
    log=open(os.path.join(outpath,filename),'w')
except Exception as e:
    print 'Could not open %s. [%s]'%( \
         os.path.join(outpath,filename), e.message)
    sys.exit(1)
#except IOError as (errno, strerror):
    #print 'Could not open %s. I/O error({0}): {1}'%( \
         #filename,errno, strerror)
    #sys.exit(1)

# write configuration information to the logfile
log.write('#PARADIGM   : %s\n'%(pfile))
log.write('#FEEDBACK   : %d\n'%(FEEDBACK))
log.write('#DETREND    : %d\n'%(DETREND))
log.write('#TCPIP      : %d\n'%(TCPIP))
log.write('#ALPHA_SCALE: %d\n'%(ALPHA_SCALE))
log.write('#ALPHA_MAX  : %d\n'%(ALPHA_MAX))
log.write('#NUM STIM   : %d\n'%(Num_Stim))
log.write('#NUM SECS   : %5.2f\n'%(Num_Secs))

# write headers of the outputs, to simplify
log.write("Time Stamp; STIM; Left Text; Right Text; Stim Text;" \
          " Show; Sign; Classifier Output; Detrended Output;" \
          " Cumulative Score\n")
log.write("Time Stamp; KEYPRESS; Key Identifier\n")

#try:
    #ser = serial.Serial(0,300,timeout=0,parity=serial.PARITY_NONE,rtscts=1)
    ##ser = serial.Serial(0,9600,timeout=0,parity=serial.PARITY_EVEN,rtscts=1)
#except:
    #sys.stderr.write("could not open serial port\n")

## activate line below for test version of this program
#serial    sys.exit(1)

#################################
#  Initialize the various bits  #
#################################

# initialize communication with the lumina
if LUMINA == 1:
    ## initialize communication with the lumina
    devices=pyxid.get_xid_devices()

    if devices:
        lumina_dev=devices[0]
    else:
        print "Could not find Lumina device"
        sys.exit(1)

    print "attached lumina device", lumina_dev
    if lumina_dev.is_response_device():
        lumina_dev.reset_base_timer()
        lumina_dev.reset_rt_timer()
    else:
        print "Error: Lumina device is not a response device??"
        log.write("Error: Lumina device is not a response device??")
        sys.exit()

# setup TCPIP communication with real time system
if TCPIP:
  from VisionEgg.TCPController import *
  tcp_server = TCPServer(hostname='127.0.0.1',
               port=TCPIP_PORT,
               single_socket_but_reconnect_ok=1,
               confirm_address_with_gui=TCPIP_GUI)
  #tcp_server = TCPServer(hostname='',
               #port=TCPIP_PORT,
               #single_socket_but_reconnect_ok=1,
               #confirm_address_with_gui=TCPIP_GUI)
  if tcp_server.server_socket is None: # User wants to quit
    sys.exit()

  tcp_listener = tcp_server.create_listener_once_connected()

# Initialize OpenGL graphics screen.
#screen = VisionEgg.Core.Screen(size=(1024,768),fullscreen=True,frameless=True)
screen = VisionEgg.Core.Screen(size=(800,600),fullscreen=False,frameless=False)
#screen = get_default_screen()

# Set the background color to white (RGBA).
screen.parameters.bgcolor = (0.0,0.0,0.0,0.0)

screen_half_x = screen.size[0]/2
screen_half_y = screen.size[1]/2

#textAA = Text(text="NO FEEDBACK",
#              color=(1.0,1.0,1.0),
#              position=(screen_half_x,screen_half_y+20),
#              font_size=60,
#              anchor='center')

fonts=pygame.font.get_fonts()
#print "I Want this Font %s\n%s\n"%\
#    (fonts[122],pygame.font.match_font(fonts[98]))

# This text box is 2/3 of the screen size wide and .5 of the scree high
# which translates to 28 characters wide and 9 characters high at 1024x768
# resolution with 40 point fixed font, some adjustments might be necessary

# concatenate string to first 252 characters
Instructions=Instructions[:252] if len(Instructions) > 252 else Instructions

textAA = WrappedText(text=Instructions,
         position=(screen.size[0]/6,screen.size[1]-screen.size[1]/4),
         size=(int((2.0/3.0)*screen.size[0]),
               int(.5*screen.size[1])),
         font_size=40,
         #font_name=pygame.font.match_font(fonts[122]),
         font_name=None,
         color=(1.0,1.0,1.0))

taskSize = 60
taskStimulus = Text(text="+",
    on    = 0,
    color=(1.0,1.0,1.0),
    position=(screen_half_x,screen_half_y - (taskSize+20)),
    font_size=taskSize,
    font_name=None,
    #font_name=pygame.font.match_font(fonts[122]),
    anchor='center'
)

centerStim = Target2D(
    on         = 0,
    anchor      = 'center',
    position    = (screen_half_x , screen_half_y),
    size        = (10, 10.0),
    color       = (1.0, 1.0, 1.0, 1.0),
    orientation = 0.0 # Up
)

target_top = 0.8*screen.size[1] + shift_unit
target_bottom = 0.2*screen.size[1] - shift_unit
stim_range_half = (target_top - target_bottom)/2.0
score_factor = 90.0 / (30.0*0.5) # assume 30 maximum TRs at maximum -.5
                                 # "score" to go from 90 to 0 deg

# Create a Viewport instance
viewport = Viewport(screen=screen, \
                    stimuli=[textAA,centerStim,taskStimulus])

# the Vision Egg's runtime control abilities.
p = Presentation( go_duration=(Num_Secs,'seconds'), \
                  trigger_go_if_armed=0, \
                  viewports=[viewport])

# calculate a few variables we need
next_stim_time = 0
direction = 1.0
last_direction = 1.0
arrow_scale = 0
arrowColor = 0
score = 90.0
first_loop = 1
start_time = 0
StimCount = -1
stimVis = 0
fbVis = 0
stimText = "+"
leftText = "left"
rightText = "right"

# some vars for receiving distance and detrending
tcp_dist = 0.0
dist = 0.0
dist_detrend= 0.0
dist_array = [];
time_array = [];
startDetrend = 0
detrendLag = 15 #TRs
currentTime = 0
currentTR = 0
currentShow = 1234 #to initialize value
currentSign = 1

#Record key presses to text file, end stimulus with 'esc'
def keydown(event):
    global currentTime

    #log the key press
    log.write("%5.3f; KEYPRESS; %s\n"%(currentTime,event.key))
    if event.key == pygame.locals.K_ESCAPE:         # Quit presentation 'p'
        p.parameters.go_duration = (0, 'frames')    # with 'ESC' key

# determine what the current state should be
def getState(t):
    global ISI, next_stim_time, direction, last_direction, arrow_scale, score
    global first_loop, start_time, StimCount, stimVis, fbVis, stimText
    global score_factor, arrowColor, leftText, rightText
    global left_strings, right_strings, stimulus_strings, show_array
    global currentTime, currentShow, currentSign, currentTR

    # CC added for TCPIP and detrends
    global tcp_dist, dist_detrend, dist_array, time_array, dist
    global startDetrend, detrendLag
    global ALPHA_MAX, ALPHA_SCALE
    global lumina_dev

    currentTime = t

    if (first_loop == 1) and (p.parameters.trigger_go_if_armed):
        first_loop = 0
        start_time = VisionEgg.time_func()
        if TCPIP:
            tcp_listener.buffer="" # reset buffer

    if t > next_stim_time:
        StimCount = StimCount + 1
        currentShow = int(np.abs(show_array[StimCount]))
        currentSign = int(np.sign(show_array[StimCount]))
        stimText = stimulus_strings[StimCount]
        leftText = left_strings[StimCount]
        rightText = right_strings[StimCount]

        next_stim_time = next_stim_time + ISI

        if StimCount > detrendLag:
            startDetrend=True

####    # note: When run with SVR testing, the input values
####    #       will range from (-inf,inf), most likely (-5,5)
    if TCPIP:
        tcp_buffer=str(tcp_listener.buffer)
        #print "%s (%s)" %('tcp_data', tcp_buffer)
        tcp_data=tcp_buffer.replace('\000', ''); # remove null byte
        #print "%s (%s) (%s)" %('tcp_data', tcp_data, tcp_buffer)

        if tcp_data != '\000' and tcp_data != "" and \
            "nan" not in tcp_data.lower():

            vals=tcp_data.split(",")
            if len(vals) > 1:
                tcp_dist=float(vals[1])
            else:
                tcp_dist=float(tcp_data)

            tcp_listener.buffer="" # reset buffer
            dist_array = dist_array + [tcp_dist]
            time_array = time_array + [t]
            if DETREND == 1:
                if startDetrend:
                    dist_detrend = detrendDist(time_array, dist_array)
                else:
                    dist_detrend = tcp_dist
                arrow_scale=dist_detrend
            else:
                arrow_scale=tcp_dist

            if currentShow == 9999 or FEEDBACK == 0:
                fbVis = 0
                score = 90.0
                arrowColor = 0
            else:
                fbVis = 1
                arrowColor = 1
                # increment the score
                score = score - currentSign*arrow_scale*score_factor
                if score < 0:
                    score = 0
                elif score > 180:
                    score = 180

            log.write("%5.3f; STIM; %s; %s; %s; %d; %d; %5.3f; %5.3f; %5.3f\n" \
                %(t, leftText, rightText, stimText, currentShow, \
                  currentSign, tcp_dist, arrow_scale, score))

    # handle anything that comes in on the button boxes
    #print "About to test LUMINA",
    #print lumina_dev
    if LUMINA == 1:
        lumina_dev.poll_for_response()
        while lumina_dev.response_queue_size() > 0:
            response = lumina_dev.get_next_response()
            if response["pressed"]:
                print "Lumina received: %s, %d"%(response["key"],response["key"])
                if response["key"] == 4:
                    log.write("%5.3f; TR; %s\n"%(currentTime,currentTR))
                    currentTR=currentTR+1
                else:
                    log.write("%5.3f; LUMINA; %d\n"%(currentTime,response["key"]))

    #print "TESTED LUMINA"
    return 1

def myRightLabel(t):
    global rightText
    return rightText

def myLeftLabel(t):
    global leftText
    return leftText

def myFixation(t):
    global stimText
    return stimText

def myArrowOn(t):
    global fbVis
    return fbVis

# define location of target
def get_line_location(t):
    global direction
    temp = (0.5*screen.size[1], 0.8*screen.size[1])
    if direction == -1.0 :
        temp = (0.5*screen.size[0], 0.2*screen.size[1])

    if direction == 1.0  :
        temp = (0.5*screen.size[0], 0.8*screen.size[1])

    return temp

# define direction of arrow
def get_arrow_direction(t):
    global score
    return score

# define size of arrow
def get_arrow_size(t):
    global score

    if arrow_scale < 1  :
        size = (score, 10)

    if arrow_scale > 0 :
        size = (score, 10)

    return size

# define size of arrow
def get_arrow_position(t):
    global score, shift_unit, screen_half_x, screen_half_y, L
    position_x = screen_half_x + L/2.0*cos(score*3.14159/180.0)
    position_y = screen_half_y + L/2.0*sin(score*3.14159/180.0)
    #position_x = screen_half_x
    #position_y = screen_half_y

    return (position_x, position_y)

#define color of arrow
def get_arrow_color(t):
    global arrowColor

    if arrowColor == 1:
        r = 1.0
        g = 0.0
        b = 0.0
        a = 1.0
    else:
        r = 1.0
        g = 1.0
        b = 1.0
        a = 1.0

    return (r,g,b,a)

def getSvmData(t):
    global dist_detrend
    return str(dist_detrend)

def detrendDist(time,dist):
    if DETREND:
        (slope,offset)=polyfit(time,dist,1)
    else:
        slope=1
        offset=0

    curr_dist = dist[-1]
    curr_time = time[-1]

    return curr_dist-(slope*curr_time+offset)

def sign(number):
    try:return number/abs(number)
    except ZeroDivisionError:return 1

def return_one(t):
    return 1

def lumina_trigger():
    global lumina_dev

    lumina_dev.poll_for_response()
    while lumina_dev.response_queue_size() > 0:
        response=lumina_dev.get_next_response()
        if response['key'] == 4:
            lumina_dev.clear_response_queue()
            print "TRIGGER RECEIVED!"
            return 1
    return 0

#######################
#  Define controllers #
#######################
###### Create an instance of the Controller class
if LUMINA == 1:
    trigger_in_controller = FunctionController(during_go_func=return_one, \
        between_go_func=lumina_trigger)
else:
    trigger_in_controller = KeyboardTriggerInController(pygame.locals.K_5)

stimulus_on_controller = ConstantController(during_go_value=1, \
                           between_go_value=0)
stimulus_off_controller = ConstantController(during_go_value=0, \
                           between_go_value=1)
state_controller = FunctionController(during_go_func=getState)
target_location_controller = FunctionController( \
                           during_go_func=get_line_location)
right_label_controller = FunctionController(during_go_func=myRightLabel)
left_label_controller = FunctionController(during_go_func=myLeftLabel)
task_controller = FunctionController(during_go_func=myFixation)
arrow_direction_controller = FunctionController( \
                           during_go_func=get_arrow_direction)

arrow_position_controller = FunctionController( \
                           during_go_func=get_arrow_position)
arrow_color_controller = FunctionController(during_go_func=get_arrow_color)
myArrowOnControl = FunctionController(during_go_func=myArrowOn)

#############################################################
#  Connect the controllers with the variables they control  #
#############################################################
# Actually listens to the TCP socket
if TCPIP: p.add_controller(None, None, tcp_listener)

p.add_controller(p,'trigger_go_if_armed',trigger_in_controller)
p.add_controller(textAA,'on', stimulus_off_controller )
#p.add_controller(textA,'on', stimulus_off_controller )
#p.add_controller(textB,'on', stimulus_off_controller )
#p.add_controller(textC,'on', stimulus_off_controller )
#p.add_controller(textD,'on', stimulus_off_controller )
p.add_controller(taskStimulus,'on', stimulus_on_controller)
p.add_controller(centerStim,'on', stimulus_on_controller)
p.add_controller(taskStimulus,'text', task_controller)
p.add_controller(p, 'trigger_go_if_armed', state_controller)
p.parameters.handle_event_callbacks = [(pygame.locals.KEYDOWN, keydown)]

#######################
#  Run the stimulus!  #
#######################

p.go()
#p.export_movie_go(frames_per_sec=0.5,filename_base='movie/frame_')
log.close
f.close

max_stars_repo_name: computational-neuroimaging-lab/mindrun

max_issues_repo_name: computational-neuroimaging-lab/mindrun

max_forks_repo_name: computational-neuroimaging-lab/mindrun

reason: from scipy


text:
import numpy as np
import pytest

from hashlib import md5
from pathlib import Path

from readimc import MCDFile

class TestMCDFile:
    damond_mcd_file_path = Path("data/Damond2019/20170814_G_SE.mcd")

    @classmethod
    def setup_class(cls):
        if cls.damond_mcd_file_path.exists():
            cls.damond_mcd_file = MCDFile(cls.damond_mcd_file_path)
            cls.damond_mcd_file.open()
        else:
            cls.damond_mcd_file = None

    @classmethod
    def teardown_class(cls):
        if cls.damond_mcd_file is not None:
            cls.damond_mcd_file.close()
            cls.damond_mcd_file = None

    def test_metadata(self, imc_test_data_mcd_file: MCDFile):
        digest = md5(imc_test_data_mcd_file.metadata.encode("utf-8")).digest()
        assert digest == b"\xac\xd8@\x0f\x0b\xf4p\x89\xdd!\xe7o\x19\xa6\x8d\x97"

    def test_slides(self, imc_test_data_mcd_file: MCDFile):
        assert len(imc_test_data_mcd_file.slides) == 1

        slide = imc_test_data_mcd_file.slides[0]
        assert slide.id == 0
        assert slide.description == "Slide"
        assert slide.width_um == 75000.0
        assert slide.height_um == 25000.0
        assert len(slide.panoramas) == 1
        assert len(slide.acquisitions) == 3

        panorama = next(p for p in slide.panoramas if p.id == 1)
        assert panorama.slide == slide
        assert panorama.id == 1
        assert panorama.description == "Panorama_001"
        assert panorama.width_um == 193.0
        assert panorama.height_um == 162.0
        assert panorama.points_um == (
            (31020.0, 13486.0),
            (31213.0, 13486.0),
            (31213.0, 13324.0),
            (31020.0, 13324.0),
        )
        assert len(panorama.acquisitions) == 0

        acquisition = next(a for a in slide.acquisitions if a.id == 1)
        assert acquisition.slide == slide
        assert acquisition.panorama is None
        assert acquisition.id == 1
        assert acquisition.description == "ROI_001"
        assert acquisition.width_px == 60
        assert acquisition.height_px == 60
        assert acquisition.pixel_size_x_um == 1.0
        assert acquisition.pixel_size_y_um == 1.0
        assert acquisition.width_um == 60.0
        assert acquisition.height_um == 60.0
        assert acquisition.num_channels == 5
        assert tuple(acquisition.channel_metals) == (
            "Ag",
            "Pr",
            "Sm",
            "Eu",
            "Yb",
        )
        assert tuple(acquisition.channel_masses) == (107, 141, 147, 153, 172)
        assert tuple(acquisition.channel_labels) == (
            "107Ag",
            "Cytoker_651((3356))Pr141",
            "Laminin_681((851))Sm147",
            "YBX1_2987((3532))Eu153",
            "H3K27Ac_1977((2242))Yb172",
        )
        assert tuple(acquisition.channel_names) == (
            "Ag107",
            "Pr141",
            "Sm147",
            "Eu153",
            "Yb172",
        )
        assert acquisition.roi_points_um == (
            (31080.0, 13449.0),
            (31140.0, 13449.0),
            (31140.0, 13389.0),
            (31080.0, 13389.0),
        )
        assert acquisition.roi_coords_um == (
            (31080.0, 13449.0),
            (31139.799043811327, 13450.084762417044),
            (31140.501, 13390.28),
            (31080.701956188677, 13389.195237582955),
        )

    def test_read_acquisition(self, imc_test_data_mcd_file: MCDFile):
        slide = imc_test_data_mcd_file.slides[0]
        acquisition = next(a for a in slide.acquisitions if a.id == 1)
        img = imc_test_data_mcd_file.read_acquisition(acquisition=acquisition)
        assert img.dtype == np.float32
        assert img.shape == (5, 60, 60)

    def test_read_slide(self, imc_test_data_mcd_file: MCDFile):
        slide = imc_test_data_mcd_file.slides[0]
        img = imc_test_data_mcd_file.read_slide(slide)
        assert img.dtype == np.uint8
        assert img.shape == (669, 2002, 4)

    def test_read_panorama(self, imc_test_data_mcd_file: MCDFile):
        slide = imc_test_data_mcd_file.slides[0]
        panorama = next(p for p in slide.panoramas if p.id == 1)
        img = imc_test_data_mcd_file.read_panorama(panorama)
        assert img.dtype == np.uint8
        assert img.shape == (162, 193, 4)

    def test_read_before_ablation_image(self, imc_test_data_mcd_file: MCDFile):
        slide = imc_test_data_mcd_file.slides[0]
        acquisition = next(a for a in slide.acquisitions if a.id == 1)
        img = imc_test_data_mcd_file.read_before_ablation_image(acquisition)
        assert img is None

    def test_read_after_ablation_image(self, imc_test_data_mcd_file: MCDFile):
        slide = imc_test_data_mcd_file.slides[0]
        acquisition = next(a for a in slide.acquisitions if a.id == 1)
        img = imc_test_data_mcd_file.read_after_ablation_image(acquisition)
        assert img is None

    @pytest.mark.skipif(not damond_mcd_file_path.exists(), reason="data not available")
    def test_metadata_damond(self, imc_test_data_mcd_file: MCDFile):
        digest = md5(imc_test_data_mcd_file.metadata.encode("utf-8")).digest()
        assert digest == b"\xac\xd8@\x0f\x0b\xf4p\x89\xdd!\xe7o\x19\xa6\x8d\x97"

    @pytest.mark.skipif(not damond_mcd_file_path.exists(), reason="data not available")
    def test_slides_damond(self):
        assert len(self.damond_mcd_file.slides) == 1

        slide = self.damond_mcd_file.slides[0]
        assert slide.id == 1
        assert slide.description == "compensationslide1000"
        assert slide.width_um == 75000.0
        assert slide.height_um == 25000.0
        assert len(slide.panoramas) == 8
        assert len(slide.acquisitions) == 41

        panorama = next(p for p in slide.panoramas if p.id == 1)
        assert panorama.slide == slide
        assert panorama.id == 1
        assert panorama.description == "TuningTape"
        assert panorama.width_um == 1472.9184890671672
        assert panorama.height_um == 1526.6011674842225
        assert panorama.points_um == (
            (28961.0, 6460.0),
            (30433.682653484964, 6486.356736527695),
            (30461.0, 4960.0),
            (28988.317346515036, 4933.643263472244),
        )
        assert len(panorama.acquisitions) == 5

        acquisition = next(a for a in panorama.acquisitions if a.id == 1)
        assert acquisition.slide == slide
        assert acquisition.panorama == panorama
        assert acquisition.id == 1
        assert acquisition.description == "TT_G01"
        assert acquisition.width_px == 51
        assert acquisition.height_px == 50
        assert acquisition.pixel_size_x_um == 1.0
        assert acquisition.pixel_size_y_um == 1.0
        assert acquisition.width_um == 51.0
        assert acquisition.height_um == 50.0
        assert acquisition.num_channels == 3
        assert tuple(acquisition.channel_metals) == ("Eu", "Eu", "Lu")
        assert tuple(acquisition.channel_masses) == (151, 153, 175)
        assert tuple(acquisition.channel_labels) == ("151Eu", "153Eu", "175Lu")
        assert tuple(acquisition.channel_names) == ("Eu151", "Eu153", "Lu175")
        assert acquisition.roi_points_um == (
            (29195.563447789347, 6091.267354770278),
            (29245.675268795687, 6091.269356041085),
            (29245.67326752488, 6041.381177047424),
            (29195.781085989598, 6041.606820330908),
        )
        assert acquisition.roi_coords_um is None

    @pytest.mark.skipif(not damond_mcd_file_path.exists(), reason="data not available")
    def test_read_acquisition_damond(self):
        slide = self.damond_mcd_file.slides[0]
        acquisition = next(a for a in slide.acquisitions if a.id == 1)
        img = self.damond_mcd_file.read_acquisition(acquisition=acquisition)
        assert img.dtype == np.float32
        assert img.shape == (3, 50, 51)

    @pytest.mark.skipif(not damond_mcd_file_path.exists(), reason="data not available")
    def test_read_slide_damond(self):
        slide = self.damond_mcd_file.slides[0]
        img = self.damond_mcd_file.read_slide(slide)
        assert img.dtype == np.uint8
        assert img.shape == (930, 2734, 3)

    @pytest.mark.skipif(not damond_mcd_file_path.exists(), reason="data not available")
    def test_read_panorama_damond(self):
        slide = self.damond_mcd_file.slides[0]
        panorama = next(p for p in slide.panoramas if p.id == 1)
        img = self.damond_mcd_file.read_panorama(panorama)
        assert img.dtype == np.uint8
        assert img.shape == (4096, 3951, 4)

    @pytest.mark.skipif(not damond_mcd_file_path.exists(), reason="data not available")
    def test_read_before_ablation_image_damond(self):
        slide = self.damond_mcd_file.slides[0]
        acquisition = next(a for a in slide.acquisitions if a.id == 1)
        img = self.damond_mcd_file.read_before_ablation_image(acquisition)
        assert img is None

    @pytest.mark.skipif(not damond_mcd_file_path.exists(), reason="data not available")
    def test_read_after_ablation_image_damond(self):
        slide = self.damond_mcd_file.slides[0]
        acquisition = next(a for a in slide.acquisitions if a.id == 1)
        img = self.damond_mcd_file.read_after_ablation_image(acquisition)
        assert img is None

max_stars_repo_name: BodenmillerGroup/readimc

max_issues_repo_name: BodenmillerGroup/readimc

max_forks_repo_name: BodenmillerGroup/readimc


text:
import numpy as np
from matplotlib import pyplot as plt
data = [1,3,3,3,3,9,9,5,4,4,8,8,8,6,7]
plt.hist(data,color = "g",bins = 4)
plt.show()


text:
import numpy as np
from PyQt5 import QtCore

class BoatSpeed(object):
    """Class to generate boat speed time series plot. If checkboxes for the boat speed reference
    (BT, GGA, VTG) are available they can be used to control what references are plotted.

    Attributes
    ----------
    canvas: MplCanvas
        Object of MplCanvas a FigureCanvas
    fig: Object
        Figure object of the canvas
    units: dict
        Dictionary of units conversions
    cb: bool
        Boolean to determine if checkboxes to control the boat speed reference are to be used
    cb_bt: QCheckBox
        Name of QCheckBox for bottom track
    cb_gga: QCheckBox
        Name of QCheckBox for GGA
    cb_vtg: QCheckBox
        Name of QCheckBox for VTG
    bt: list
        Plot reference for bottom track
    gga: list
        Plot reference for GGA
    vtg: list
        Plot reference for VTG
    hover_connection: int
        Index to data cursor connection
    annot: Annotation
        Annotation object for data cursor
    """

    def __init__(self, canvas):
        """Initialize object using the specified canvas.

        Parameters
        ----------
        canvas: MplCanvas
            Object of MplCanvas
        """

        # Initialize attributes
        self.canvas = canvas
        self.fig = canvas.fig
        self.units = None
        self.cb = None
        self.cb_bt = None
        self.cb_gga = None
        self.cb_vtg = None
        self.bt = None
        self.gga = None
        self.vtg = None
        self.hover_connection = None
        self.annot = None

    def create(self, transect, units,
               cb=False, cb_bt=None, cb_gga=None, cb_vtg=None):
        """Create the axes and lines for the figure.

        Parameters
        ----------
        transect: TransectData
            Object of TransectData containing boat speeds to be plotted
        units: dict
            Dictionary of units conversions
        cb: bool
            Boolean to determine if checkboxes to control the boat speed reference are to be used
        cb_bt: QCheckBox
            Name of QCheckBox for bottom track
        cb_gga: QCheckBox
            Name of QCheckBox for GGA
        cb_vtg: QCheckBox
            Name of QCheckBox for VTG
        """

        # Assign and save parameters
        self.units = units
        self.cb = cb
        self.cb_bt = cb_bt
        self.cb_gga = cb_gga
        self.cb_vtg = cb_vtg

        # Check the checkbox to determine what should be shown in the plot
        control = self.checkbox_control(transect)

        # Clear the plot
        self.fig.clear()

        # Configure axis
        self.fig.ax = self.fig.add_subplot(1, 1, 1)

        # Set margins and padding for figure
        self.fig.subplots_adjust(left=0.08, bottom=0.2, right=0.98, top=0.98, wspace=0.1, hspace=0)
        self.fig.ax.set_xlabel(self.canvas.tr('Ensembles'))
        self.fig.ax.set_ylabel(self.canvas.tr('Boat speed' + units['label_V']))
        self.fig.ax.grid()
        self.fig.ax.xaxis.label.set_fontsize(12)
        self.fig.ax.yaxis.label.set_fontsize(12)
        self.fig.ax.tick_params(axis='both', direction='in', bottom=True, top=True, left=True, right=True)

        # Initialize max trackers
        max_gga = np.nan
        max_vtg = np.nan

        ensembles = np.arange(1, len(transect.boat_vel.bt_vel.u_mps) + 1)

        # Plot bottom track boat speed
        speed = np.sqrt(transect.boat_vel.bt_vel.u_processed_mps ** 2 + transect.boat_vel.bt_vel.v_processed_mps ** 2)
        self.bt = self.fig.ax.plot(ensembles, speed * units['V'], 'r-')

        # Plot invalid data points using a symbol to represent what caused the data to be invalid
        invalid_bt = np.logical_not(transect.boat_vel.bt_vel.valid_data)
        if invalid_bt is not None:
            speed = np.sqrt(
                transect.boat_vel.bt_vel.u_mps ** 2 + transect.boat_vel.bt_vel.v_mps ** 2)
            speed[np.isnan(speed)] = 0
            self.bt.append(self.fig.ax.plot(ensembles[invalid_bt[1]], speed[invalid_bt[1]] * units['V'],
                                            'k', linestyle='', marker='$O$')[0])
            self.bt.append(self.fig.ax.plot(ensembles[invalid_bt[2]], speed[invalid_bt[2]] * units['V'],
                                            'k', linestyle='', marker='$E$')[0])
            self.bt.append(self.fig.ax.plot(ensembles[invalid_bt[3]], speed[invalid_bt[3]] * units['V'],
                                            'k', linestyle='', marker='$V$')[0])
            self.bt.append(self.fig.ax.plot(ensembles[invalid_bt[4]], speed[invalid_bt[4]] * units['V'],
                                            'k', linestyle='', marker='$S$')[0])
            self.bt.append(self.fig.ax.plot(ensembles[invalid_bt[5]], speed[invalid_bt[5]] * units['V'],
                                            'k', linestyle='', marker='$B$')[0])

        max_bt = np.nanmax(speed)

        # Based on checkbox control make bt visible or not
        if control['bt']:
            for item in self.bt:
                item.set_visible(True)
        else:
            for item in self.bt:
                item.set_visible(False)

        # Plot VTG boat speed
        if transect.boat_vel.vtg_vel is not None:
            speed = np.sqrt(
                transect.boat_vel.vtg_vel.u_processed_mps ** 2 + transect.boat_vel.vtg_vel.v_processed_mps ** 2)
            self.vtg = self.fig.ax.plot(ensembles, speed * units['V'], 'g-')

            # Plot invalid data points using a symbol to represent what caused the data to be invalid
            invalid_gps = np.logical_not(transect.boat_vel.vtg_vel.valid_data)
            # if invalid_gps is not None:
            if 0 < np.sum(invalid_gps[0, :]) < invalid_gps.shape[1]:
                speed = np.sqrt(
                    transect.boat_vel.vtg_vel.u_mps ** 2 + transect.boat_vel.vtg_vel.v_mps ** 2)
                speed[np.isnan(speed)] = 0
                self.vtg.append(self.fig.ax.plot(ensembles[invalid_gps[1]], speed[invalid_gps[1]] * units['V'],
                                                 'k', linestyle='', marker='$O$')[0])
                self.vtg.append(self.fig.ax.plot(ensembles[invalid_gps[5]], speed[invalid_gps[5]] * units['V'],
                                                 'k', linestyle='', marker='$H$')[0])
                self.vtg.append(self.fig.ax.plot(ensembles[invalid_gps[4]], speed[invalid_gps[4]] * units['V'],
                                                 'k', linestyle='', marker='$S$')[0])

            max_vtg = np.nanmax(speed)
            if control['vtg']:
                for item in self.vtg:
                    item.set_visible(True)
            else:
                for item in self.vtg:
                    item.set_visible(False)

        # Plot GGA boat speed
        if transect.boat_vel.gga_vel is not None:
            speed = np.sqrt(
                transect.boat_vel.gga_vel.u_processed_mps ** 2 + transect.boat_vel.gga_vel.v_processed_mps ** 2)
            self.gga = self.fig.ax.plot(ensembles, speed * units['V'], 'b-')

            # Plot invalid data points using a symbol to represent what caused the data to be invalid
            invalid_gps = np.logical_not(transect.boat_vel.gga_vel.valid_data)
            if 0 < np.sum(invalid_gps[0, :]) < invalid_gps.shape[1]:
                speed = np.sqrt(
                    transect.boat_vel.gga_vel.u_mps ** 2 + transect.boat_vel.gga_vel.v_mps ** 2)
                speed[np.isnan(speed)] = 0
                self.gga.append(self.fig.ax.plot(ensembles[invalid_gps[1]], speed[invalid_gps[1]] * units['V'],
                                                 'k', linestyle='', marker='$O$')[0])
                self.gga.append(self.fig.ax.plot(ensembles[invalid_gps[2]], speed[invalid_gps[2]] * units['V'],
                                                 'k', linestyle='', marker='$Q$')[0])
                self.gga.append(self.fig.ax.plot(ensembles[invalid_gps[3]], speed[invalid_gps[3]] * units['V'],
                                                 'k', linestyle='', marker='$A$')[0])
                self.gga.append(self.fig.ax.plot(ensembles[invalid_gps[5]], speed[invalid_gps[5]] * units['V'],
                                                 'k', linestyle='', marker='$H$')[0])
                self.gga.append(self.fig.ax.plot(ensembles[invalid_gps[4]], speed[invalid_gps[4]] * units['V'],
                                                 'k', linestyle='', marker='$S$')[0])

            max_gga = np.nanmax(speed)
            if control['gga']:
                for item in self.gga:
                    item.set_visible(True)
            else:
                for item in self.gga:
                    item.set_visible(False)

        # Set axis limits
        max_y = np.nanmax([max_bt, max_gga, max_vtg]) * 1.1
        self.fig.ax.set_ylim(top=np.ceil(max_y * units['L']), bottom=-0.5)
        self.fig.ax.set_xlim(left=-1 * ensembles[-1] * 0.02, right=ensembles[-1] * 1.02)

        if transect.start_edge == 'Right':
            self.fig.ax.invert_xaxis()
            self.fig.ax.set_xlim(right=-1 * ensembles[-1] * 0.02, left=ensembles[-1] * 1.02)

        # Initialize annotation for data cursor
        self.annot = self.fig.ax.annotate("", xy=(0, 0), xytext=(-20, 20), textcoords="offset points",
                                          bbox=dict(boxstyle="round", fc="w"),
                                          arrowprops=dict(arrowstyle="->"))

        self.annot.set_visible(False)

        self.canvas.draw()

    def checkbox_control(self, transect):
        """Controls the visibility and checks the status of checkboxes.

        Parameters
        ----------
        transect: TransectData
            Object of TransectData with boat speeds to be plotted.
        """

        # Initialize control dictionary
        control = {'bt': True, 'gga': True, 'vtg': True, 'vectors': True}

        # If checkboxes are available, enable the checkboxes if transect contains that type of data
        if self.cb:
            # Enable check boxes as data is available
            if transect.boat_vel.gga_vel is not None:
                self.cb_gga.setEnabled(True)
            else:
                self.cb_gga.setCheckState(QtCore.Qt.Unchecked)
                self.cb_gga.setEnabled(False)

            if transect.boat_vel.vtg_vel is not None:
                self.cb_vtg.setEnabled(True)
            else:
                self.cb_vtg.setCheckState(QtCore.Qt.Unchecked)
                self.cb_vtg.setEnabled(False)

            # Get checkbox status
            # BT
            if self.cb_bt.checkState() == QtCore.Qt.Checked:
                control['bt'] = True
            else:
                control['bt'] = False
            # GGA
            if self.cb_gga.checkState() == QtCore.Qt.Checked:
                control['gga'] = True
            else:
                control['gga'] = False
            # VTG
            if self.cb_vtg.checkState() == QtCore.Qt.Checked:
                control['vtg'] = True
            else:
                control['vtg'] = False
        return control

    def change(self):
        """Changes the visibility of the various boat speed references based on user input via checkboxes.
        """

        # If checkboxes are available, check status and set boat speed reference line visibility accordingly.
        if self.cb:
            if self.cb_bt.checkState() == QtCore.Qt.Checked:
                for item in self.bt:
                    item.set_visible(True)
            else:
                for item in self.bt:
                    item.set_visible(False)
            # GGA
            if self.cb_gga.checkState() == QtCore.Qt.Checked:
                for item in self.gga:
                    item.set_visible(True)
                # self.gga[0].set_visible(True)
            elif self.gga is not None:
                for item in self.gga:
                    item.set_visible(False)
                # self.gga[0].set_visible(False)
            # VTG
            if self.cb_vtg.checkState() == QtCore.Qt.Checked:
                for item in self.vtg:
                    item.set_visible(True)
                # self.vtg[0].set_visible(True)
            elif self.vtg is not None:
                for item in self.vtg:
                    item.set_visible(False)
                # self.vtg[0].set_visible(False)

            # Draw canvas
            self.canvas.draw()

    def update_annot(self, ind, plt_ref, ref_label):
        """Updates the location and text and makes visible the previously initialized and hidden annotation.

        Parameters
        ----------
        ind: dict
            Contains data selected.
        plt_ref: Line2D
            Reference containing plotted data
        ref_label: str
            Label used to ID data type in annotation
        """

        # Get selected data coordinates
        pos = plt_ref._xy[ind["ind"][0]]

        # Shift annotation box left or right depending on which half of the axis the pos x is located and the
        # direction of x increasing.
        if plt_ref.axes.viewLim.intervalx[0] < plt_ref.axes.viewLim.intervalx[1]:
            if pos[0] < (plt_ref.axes.viewLim.intervalx[0] + plt_ref.axes.viewLim.intervalx[1]) / 2:
                self.annot._x = -20
            else:
                self.annot._x = -80
        else:
            if pos[0] < (plt_ref.axes.viewLim.intervalx[0] + plt_ref.axes.viewLim.intervalx[1]) / 2:
                self.annot._x = -80
            else:
                self.annot._x = -20

        # Shift annotation box up or down depending on which half of the axis the pos y is located and the
        # direction of y increasing.
        if plt_ref.axes.viewLim.intervaly[0] < plt_ref.axes.viewLim.intervaly[1]:
            if pos[1] > (plt_ref.axes.viewLim.intervaly[0] + plt_ref.axes.viewLim.intervaly[1]) / 2:
                self.annot._y = -40
            else:
                self.annot._y = 20
        else:
            if pos[1] > (plt_ref.axes.viewLim.intervaly[0] + plt_ref.axes.viewLim.intervaly[1]) / 2:
                self.annot._y = 20
            else:
                self.annot._y = -40

        self.annot.xy = pos

        # Format and display text
        text = 'x: {:.2f}, {}: {:.2f}'.format(pos[0], ref_label, pos[1])
        self.annot.set_text(text)

    def hover(self, event):
        """Determines if the user has selected a location with data and makes
        annotation visible and calls method to update the text of the annotation. If the
        location is not valid the existing annotation is hidden.

        Parameters
        ----------
        event: MouseEvent
            Triggered when mouse button is pressed.
        """

        # Set annotation to visible
        vis = self.annot.get_visible()

        # Determine if mouse location references a data point in the plot and update the annotation.
        if event.inaxes == self.fig.ax:
            cont_bt = False
            cont_gga = False
            cont_vtg = False
            ind_bt = None
            ind_gga = None
            ind_vtg = None

            if self.bt is not None:
                cont_bt, ind_bt = self.bt[0].contains(event)
            if self.gga is not None:
                cont_gga, ind_gga = self.gga[0].contains(event)
            if self.vtg is not None:
                cont_vtg, ind_vtg = self.vtg[0].contains(event)
            if cont_bt and self.bt[0].get_visible():
                self.update_annot(ind_bt, self.bt[0], 'BT')
                self.annot.set_visible(True)
                self.canvas.draw_idle()
            elif cont_gga and self.gga[0].get_visible():
                self.update_annot(ind_gga, self.gga[0], 'GGA')
                self.annot.set_visible(True)
                self.canvas.draw_idle()
            elif cont_vtg and self.vtg[0].get_visible():
                self.update_annot(ind_vtg, self.vtg[0], 'VTG')
                self.annot.set_visible(True)
                self.canvas.draw_idle()
            else:
                # If the cursor location is not associated with the plotted data hide the annotation.
                if vis:
                    self.annot.set_visible(False)
                    self.canvas.draw_idle()

    def set_hover_connection(self, setting):
        """Turns the connection to the mouse event on or off.

        Parameters
        ----------
        setting: bool
            Boolean to specify whether the connection for the mouse event is active or not.
        """

        if setting and self.hover_connection is None:
            self.hover_connection = self.canvas.mpl_connect('button_press_event', self.hover)
        elif not setting:
            self.canvas.mpl_disconnect(self.hover_connection)
            self.hover_connection = None
            self.annot.set_visible(False)
            self.canvas.draw_idle()

max_stars_repo_name: EstebM/QRevPy

max_issues_repo_name: EstebM/QRevPy

max_forks_repo_name: EstebM/QRevPy


text:
import glob
import os
import json
import time

import imageio
import pandas as pd
import numpy as np
import plotly.express as px

def xyxy_to_xywh(rect):
    w = rect["x1"] - rect["x0"]
    h = rect["y1"] - rect["y0"]
    return rect["x0"], rect["y0"], w, h

def create_annotation(annot, rectangles):
    if len(rectangles) != len(annot):
        raise Exception("All rectangles need to have an annotation.")

    x = np.array(list(map(lambda x: x["x0"], rectangles)))
    sorted_args = np.argsort(x)

    return [{"rect": xyxy_to_xywh(rectangles[i]), "class":annot[i]} for i in sorted_args]

class Annotator():
    def __init__(self):
        self.cur_img = None
        self.cur_img_path = None
        self.cur_img_num = None
        self.image_fnames = None
        self.last_shapes = None
        self.annotations = None
        self.project_dir = None
        self.t1_speed = None
        self.anno_num = None

    def store(self):
        if self.project_dir is None:
            return
        conf = {}
        conf["cur_img_path"] = self.cur_img_path
        conf["cur_img_num"] = self.cur_img_num
        conf["image_fnames"] = self.image_fnames
        conf["last_shapes"] = self.last_shapes
        conf["annotations"] = self.annotations
        conf["project_dir"] = self.project_dir

        jname = "annotator_conf.json"
        with open(os.path.join(self.project_dir, jname), "w") as fson:
            json.dump(conf, fson)

    @staticmethod
    def reload(fdir):
        fname = os.path.join(fdir, "annotator_conf.json")
        if os.path.exists(fname):
            with open(fname, "r") as fson:
                conf = json.load(fson)
            a = Annotator()
            a.cur_img_path = conf["cur_img_path"]
            a.cur_img_num = conf["cur_img_num"]
            a.image_fnames = conf["image_fnames"]
            a.last_shapes = conf["last_shapes"]
            a.annotations = conf["annotations"]
            a.project_dir = conf["project_dir"]
            a.cur_img = imageio.imread(a.cur_img_path)
            a.reset_clock()
            return a
        else:
            return None

    def load_images(self, fdir, img_types):
        if not os.path.exists(fdir):
            raise Exception("Path is not valid: ", fdir)

        self.image_fnames = glob.glob(os.path.join(fdir, img_types))
        self.image_fnames.sort(key=os.path.getmtime)
        self.annotations = {}
        self.project_dir = fdir
        self.reset_clock()

    def load_next_img(self):
        if self.image_fnames is None:
            raise Exception("No images to load. Please load file dir first.")

        if self.cur_img_path is None:
            self.cur_img_num = 0
        else:
            self.cur_img_num += 1

        self.cur_img_path = str(self.image_fnames[self.cur_img_num])
        self.cur_img = imageio.imread(self.cur_img_path)

    def plot_current_img(self):
        if self.cur_img is None:
            raise Exception("Please load img first.")

        fig = px.imshow(self.cur_img, color_continuous_scale='gray')
        # Define dragmode, newshape parameters, amd add modebar buttons
        fig.update_layout(
            dragmode='drawrect', # define dragmode
            newshape=dict(line_color='cyan'))
        if self.last_shapes is not None:
            fig.update_layout(shapes=self.last_shapes)
        return fig

    def annotate_cur_img(self, annot, shapes):
        rects = []
        for shape in shapes:
            if shape["type"] != "rect":
                continue
            rects.append(shape)
        self.annotations[self.cur_img_path] = create_annotation(annot, rects)
        self.last_shapes = shapes
        self.anno_num += 1

    def ignore_img(self):
        self.annotations[self.cur_img_path] = []
        self.anno_num += 1

    def reset_clock(self):
        self.anno_num = 0
        self.t1_speed = time.time()

    def calc_speed_end(self):
        if self.t1_speed is None or self.anno_num is None:
            raise Exception("Reset the clock first.")

        if self.anno_num == 0:
            raise Exception("Please annotate something.")

        t2 = time.time()
        speed = (t2 - self.t1_speed)/self.anno_num
        to_go = len(self.image_fnames) - len(self.annotations)
        end = to_go*speed
        return speed, end, to_go


text:
import functools

import numpy

def _fp16_mixed_precision_helper(fn):
    """Decorator to perform computation in FP32 for FP16 inputs/outputs

    Decorator to perform forward computation in FP32 for FP16 inputs,
    returning outputs casted back to FP16. Do nothing for FP32 and FP64
    inputs.
    """
    @functools.wraps(fn)
    def wrapper(self, in_data):
        flag = all([x.dtype.kind != 'f' or x.dtype == numpy.float16
                    for x in in_data])

        in_data1 = []
        for x in in_data:
            if x.dtype == numpy.float16:
                in_data1.append(x.astype(numpy.float32))
            else:
                in_data1.append(x)
        in_data1 = tuple(in_data1)

        out_data = fn(self, in_data1)

        if flag:
            out_data1 = []
            for y in out_data:
                if y is not None and y.dtype == numpy.float32:
                    out_data1.append(y.astype(numpy.float16))
                else:
                    out_data1.append(y)
            out_data = tuple(out_data1)

        return out_data
    return wrapper

max_stars_repo_name: zaltoprofen/chainer

max_issues_repo_name: nolfwin/chainer

max_forks_repo_name: nolfwin/chainer