text:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Module holding the bi-exponential model for fitting."""
import numpy as np
from pyqmri.models.template import BaseModel, constraints

class Model(BaseModel):
    """Bi-exponential model for MRI parameter quantification.

    This class holds a bi-exponential model for fitting complex MRI data.
    It realizes a forward application of the analytical signal expression
    and the partial derivatives with respesct to each parameter of interest,
    as required by the abstract methods in the BaseModel.

    Parameters
    ----------
      par : dict
        A python dict containing the necessary information to
        setup the object. Needs to contain the sequence related parametrs,
        e.g. TR, TE, TI, to fully describe the acquisitio process

    Attributes
    ----------
      TE : float
        Echo time (or any other timing valid in a bi-exponential fit).
      uk_scale : list of float
        Scaling factors for each unknown to balance the partial derivatives.
      guess : numpy.array, None
        The initial guess. Needs to be set using "computeInitialGuess"
        prior to fitting.
    """

    def __init__(self, par):
        super().__init__(par)
        self.TE = np.ones((self.NScan, 1, 1, 1))

        for i in range(self.NScan):
            self.TE[i, ...] = par["TE"][i] * np.ones((1, 1, 1))

        par["unknowns_TGV"] = 5
        par["unknowns_H1"] = 0
        par["unknowns"] = par["unknowns_TGV"]+par["unknowns_H1"]

        for j in range(par["unknowns"]):
            self.uk_scale.append(1)

        self.constraints.append(
            constraints(
                -100 / self.uk_scale[0],
                100 / self.uk_scale[0], False))
        self.constraints.append(
            constraints(
                0 / self.uk_scale[1],
                1 / self.uk_scale[1],
                True))
        self.constraints.append(
            constraints(
                ((1 / 150) / self.uk_scale[2]),
                ((1 / 1e-4) / self.uk_scale[2]),
                True))
        self.constraints.append(
            constraints(
                0 / self.uk_scale[3],
                1 / self.uk_scale[3],
                True))
        self.constraints.append(
            constraints(
                ((1 / 1500) / self.uk_scale[4]),
                ((1 / 150) / self.uk_scale[4]),
                True))
        self.guess = None

        self._ax = []
        self._M0_plot = None
        self._M0_plot_cor = None
        self._M0_plot_sag = None

        self._M01_plot = None
        self._M01_plot_cor = None
        self._M01_plot_sag = None

        self._M02_plot = None
        self._M02_plot_cor = None
        self._M02_plot_sag = None

        self._T21_plot = None
        self._T21_plot_cor = None
        self._T21_plot_sag = None

        self._T22_plot = None
        self._T22_plot_cor = None
        self._T22_plot_sag = None

    def rescale(self, x):
        """Rescale the unknowns with the scaling factors.

        Rescales each unknown with the corresponding scaling factor and
        applies a 1/x transformation for the time constants of the
        exponentials, yielding a result in milliseconds.

        Parameters
        ----------
          x : numpy.array
            The array of unknowns to be rescaled

        Returns
        -------
          numpy.array:
            The rescaled unknowns
        """
        tmp_x = np.copy(x)
        tmp_x[0] = x[0, ...] * self.uk_scale[0]
        tmp_x[1] = x[1, ...] * self.uk_scale[1]
        tmp_x[2] = 1 / (x[2, ...] * self.uk_scale[2])
        tmp_x[3] = x[3, ...] * self.uk_scale[3]
        tmp_x[4] = 1 / (x[4, ...] * self.uk_scale[4])

        const = []
        for constrained in self.constraints:
            const.append(constrained.real)
        return {"data": tmp_x,
                "unknown_name": ["M0", "M01", "T21",
                                 "M02", "T22"],
                "real_valued": const}

    def _execute_forward_3D(self, x):
        M0 = x[0, ...] * self.uk_scale[0]
        M01 = x[1, ...] * self.uk_scale[1]
        T21 = x[2, ...] * self.uk_scale[2]
        M02 = x[3, ...] * self.uk_scale[3]
        T22 = x[4, ...] * self.uk_scale[4]
        S = M0 * (M01 * np.exp(-self.TE * (T21)) +
                  M02 * np.exp(-self.TE * (T22)))
        S[~np.isfinite(S)] = 1e-20
        S = np.array(S, dtype=self._DTYPE)
        return S

    def _execute_gradient_3D(self, x):
        M0 = x[0, ...]
        M01 = x[1, ...]
        T21 = x[2, ...]
        M02 = x[3, ...]
        T22 = x[4, ...]
        grad_M0 = self.uk_scale[0] * (
            M01 * self.uk_scale[1] * np.exp(-self.TE * (
                T21 * self.uk_scale[2])) +
            M02 * self.uk_scale[3] *
            np.exp(-self.TE * (T22 * self.uk_scale[4])))
        grad_M01 = self.uk_scale[0] * M0 * self.uk_scale[1] * \
            np.exp(-self.TE * (T21 * self.uk_scale[2]))
        grad_T21 = -self.uk_scale[0] * M0 * M01 * self.uk_scale[1] * \
            self.TE * \
            self.uk_scale[2] * np.exp(-self.TE * (T21 * self.uk_scale[2]))
        grad_M02 = self.uk_scale[0] * M0 * self.uk_scale[3] * \
            np.exp(-self.TE * (T22 * self.uk_scale[4]))
        grad_T22 = -self.uk_scale[0] * M0 * M02 * self.uk_scale[3] * \
            self.TE * \
            self.uk_scale[4] * np.exp(-self.TE * (T22 * self.uk_scale[4]))
        grad = np.array([grad_M0, grad_M01, grad_T21,
                         grad_M02, grad_T22], dtype=self._DTYPE)
        grad[~np.isfinite(grad)] = 1e-20
        return grad

    def computeInitialGuess(self, *args):
        """Initialize unknown array for the fitting.

        This function provides an initial guess for the fitting.

        Parameters
        ----------
          args : list of objects
            Serves as universal interface. No objects need to be passed
            here.
        """
        test_M0 = 0.1 * np.ones(
            (self.NSlice, self.dimY, self.dimX),
            dtype=self._DTYPE)
        test_M01 = 0.5 * np.ones(
            (self.NSlice, self.dimY, self.dimX), dtype=self._DTYPE)
        test_T11 = 1/10 * np.ones(
            (self.NSlice, self.dimY, self.dimX), dtype=self._DTYPE)
        test_T12 = 1/150 * np.ones(
            (self.NSlice, self.dimY, self.dimX), dtype=self._DTYPE)

        self.guess = np.array([
            test_M0,
            test_M01,
            test_T11,
            test_M01,
            test_T12], dtype=self._DTYPE)

max_stars_repo_name: agahkarakuzu/PyQMRI

max_issues_repo_name: agahkarakuzu/PyQMRI

max_forks_repo_name: agahkarakuzu/PyQMRI


text:
import csv
from os import listdir

import numpy as np

def dummy_data_generator(backcast_length, forecast_length, signal_type='seasonality', random=False, batch_size=32):
    def get_x_y():
        lin_space = np.linspace(-backcast_length, forecast_length, backcast_length + forecast_length)
        if random:
            offset = np.random.standard_normal() * 0.1
        else:
            offset = 1
        if signal_type == 'trend':
            a = lin_space + offset
        elif signal_type == 'seasonality':
            a = np.cos(2 * np.random.randint(low=1, high=3) * np.pi * lin_space)
            a += np.cos(2 * np.random.randint(low=2, high=4) * np.pi * lin_space)
            a += lin_space * offset + np.random.rand() * 0.1
        elif signal_type == 'cos':
            a = np.cos(2 * np.pi * lin_space)
        else:
            raise Exception('Unknown signal type.')

        x = a[:backcast_length]
        y = a[backcast_length:]

        min_x, max_x = np.minimum(np.min(x), 0), np.max(np.abs(x))

        x -= min_x
        y -= min_x

        x /= max_x
        y /= max_x

        return x, y

    def gen():
        while True:
            xx = []
            yy = []
            for i in range(batch_size):
                x, y = get_x_y()
                xx.append(x)
                yy.append(y)
            yield np.array(xx), np.array(yy)

    return gen()

def get_m4_data(backcast_length, forecast_length, is_training=True):
    # https://www.mcompetitions.unic.ac.cy/the-dataset/

    if is_training:
        filename = 'data/m4/train/Daily-train.csv'
    else:
        filename = 'data/m4/val/Daily-test.csv'

    x = np.array([]).reshape(0, backcast_length)
    y = np.array([]).reshape(0, forecast_length)
    x_tl = []
    headers = True
    with open(filename, "r") as file:
        reader = csv.reader(file, delimiter=',')
        for line in reader:
            line = line[1:]
            if not headers:
                x_tl.append(line)
            if headers:
                headers = False
    x_tl_tl = np.array(x_tl)
    for i in range(x_tl_tl.shape[0]):
        if len(x_tl_tl[i]) < backcast_length + forecast_length:
            continue
        time_series = np.array(x_tl_tl[i])
        time_series = [float(s) for s in time_series if s != '']
        time_series_cleaned = np.array(time_series)
        if is_training:
            time_series_cleaned_forlearning_x = np.zeros((1, backcast_length))
            time_series_cleaned_forlearning_y = np.zeros((1, forecast_length))
            j = np.random.randint(backcast_length, time_series_cleaned.shape[0] + 1 - forecast_length)
            time_series_cleaned_forlearning_x[0, :] = time_series_cleaned[j - backcast_length: j]
            time_series_cleaned_forlearning_y[0, :] = time_series_cleaned[j:j + forecast_length]
        else:
            time_series_cleaned_forlearning_x = np.zeros(
                (time_series_cleaned.shape[0] + 1 - (backcast_length + forecast_length), backcast_length))
            time_series_cleaned_forlearning_y = np.zeros(
                (time_series_cleaned.shape[0] + 1 - (backcast_length + forecast_length), forecast_length))
            for j in range(backcast_length, time_series_cleaned.shape[0] + 1 - forecast_length):
                time_series_cleaned_forlearning_x[j - backcast_length, :] = time_series_cleaned[j - backcast_length:j]
                time_series_cleaned_forlearning_y[j - backcast_length, :] = time_series_cleaned[j: j + forecast_length]
        x = np.vstack((x, time_series_cleaned_forlearning_x))
        y = np.vstack((y, time_series_cleaned_forlearning_y))

    return x, y

def dummy_data_generator_multivariate(backcast_length, forecast_length, signal_type='seasonality', random=False,
                                      batch_size=32):
    def get_x_y():
        lin_space = np.linspace(-backcast_length, forecast_length, backcast_length + forecast_length)
        if random:
            offset = np.random.standard_normal() * 0.1
        else:
            offset = 1
        if signal_type == 'trend':
            a = lin_space + offset
        elif signal_type == 'seasonality':
            a = np.cos(2 * np.random.randint(low=1, high=3) * np.pi * lin_space)
            a += np.cos(2 * np.random.randint(low=2, high=4) * np.pi * lin_space)
            a += lin_space * offset + np.random.rand() * 0.1
        elif signal_type == 'cos':
            a = np.cos(2 * np.pi * lin_space)
        else:
            raise Exception('Unknown signal type.')

        x = a[:backcast_length]
        y = a[backcast_length:]

        min_x, max_x = np.minimum(np.min(x), 0), np.max(np.abs(x))

        x -= min_x
        y -= min_x

        x /= max_x
        y /= max_x

        return x, y

    def gen():
        while True:
            xx = []
            yy = []
            for i in range(batch_size):
                x, y = get_x_y()
                xx.append(x)
                yy.append(y)
            yield np.array(xx), np.array(yy)

    return gen()

def get_m4_data_multivariate(backcast_length, forecast_length, is_training=True):
    # to be downloaded from https://www.mcompetitions.unic.ac.cy/the-dataset/

    filename = '../examples/data/m4/train/Daily-train.csv'
    x_tl = []
    x_max = []
    headers = True
    with open(filename, "r") as file:
        reader = csv.reader(file, delimiter=',')
        for line in reader:
            line = line[1:]
            if not headers:
                x_tl.append(line)
            if headers:
                headers = False
    x_tl_tl = np.array(x_tl)
    for i in range(x_tl_tl.shape[0]):
        if len(x_tl_tl[i]) < backcast_length + forecast_length:
            continue
        time_series = np.array(x_tl_tl[i])
        time_series = [float(s) for s in time_series if s != '']
        x_max.append(np.max(time_series))
    x_max = np.max(x_max)

    if is_training:
        filename = '../examples/data/m4/train/Daily-train.csv'
    else:
        filename = '../examples/data/m4/val/Daily-test.csv'

    x = np.array([]).reshape(0, backcast_length)
    y = np.array([]).reshape(0, forecast_length)
    x_tl = []
    headers = True
    with open(filename, "r") as file:
        reader = csv.reader(file, delimiter=',')
        for line in reader:
            line = line[1:]
            if not headers:
                x_tl.append(line)
            if headers:
                headers = False
    x_tl_tl = np.array(x_tl)
    for i in range(x_tl_tl.shape[0]):
        if len(x_tl_tl[i]) < backcast_length + forecast_length:
            continue
        time_series = np.array(x_tl_tl[i])
        time_series = [float(s) for s in time_series if s != '']
        time_series = time_series / x_max
        if is_training:
            time_series_cleaned_forlearning_x = np.zeros((1, backcast_length))
            time_series_cleaned_forlearning_y = np.zeros((1, forecast_length))
            j = np.random.randint(backcast_length, time_series.shape[0] + 1 - forecast_length)
            time_series_cleaned_forlearning_x[0, :] = time_series[j - backcast_length: j]
            time_series_cleaned_forlearning_y[0, :] = time_series[j:j + forecast_length]
        else:
            time_series_cleaned_forlearning_x = np.zeros(
                (time_series.shape[0] + 1 - (backcast_length + forecast_length), backcast_length))
            time_series_cleaned_forlearning_y = np.zeros(
                (time_series.shape[0] + 1 - (backcast_length + forecast_length), forecast_length))
            for j in range(backcast_length, time_series.shape[0] + 1 - forecast_length):
                time_series_cleaned_forlearning_x[j - backcast_length, :] = time_series[j - backcast_length:j]
                time_series_cleaned_forlearning_y[j - backcast_length, :] = time_series[j: j + forecast_length]
        x = np.vstack((x, time_series_cleaned_forlearning_x))
        y = np.vstack((y, time_series_cleaned_forlearning_y))

    return x.reshape(x.shape[0], x.shape[1], 1), None, y.reshape(y.shape[0], y.shape[1], 1)

def process_data(filename):
    import wfdb
    ecg_list = listdir(filename)
    sample_list = [ecg[:-4] for ecg in ecg_list]
    clean_sample_list = [ecg for ecg in sample_list if
                         ecg not in ['102-0', 'ANNOTA', 'REC', 'SHA256SUMS', 'mitd', 'x_m']]
    all_samples = np.zeros((len(clean_sample_list), 650000, 2))
    for idx, ecg in enumerate(clean_sample_list):
        record = wfdb.rdrecord(filename + ecg)
        all_samples[idx] = record.p_signal

    return all_samples

def get_kcg_data(backcast_length, forecast_length, is_training=True):
    # to be downloaded from https://physionet.org/content/mitdb/1.0.0/
    # once downloaded should be put in ../examples/data/kcg/

    dataset = process_data(filename='../examples/data/kcg/')
    x_max = np.amax(np.abs(dataset[:195, :, :]), axis=(0, 1))

    if is_training:
        dataset = dataset[:195, :, :]
    else:
        dataset = dataset[195:, 30000:30000 + backcast_length + forecast_length + 10, :]

    x = np.array([]).reshape(0, backcast_length, 2)
    y = np.array([]).reshape(0, forecast_length, 2)

    for i in range(dataset.shape[0]):
        if (dataset[i].shape[0] < backcast_length + forecast_length):
            continue
        time_series = dataset[i]
        time_series = time_series / x_max
        if is_training:
            time_series_cleaned_forlearning_x = np.zeros((1, backcast_length, 2))
            time_series_cleaned_forlearning_y = np.zeros((1, forecast_length, 2))
            j = np.random.randint(backcast_length, time_series.shape[0] + 1 - forecast_length)
            time_series_cleaned_forlearning_x[0] = time_series[j - backcast_length: j, :]
            time_series_cleaned_forlearning_y[0] = time_series[j:j + forecast_length, :]
        else:
            time_series_cleaned_forlearning_x = np.zeros(
                (time_series.shape[0] + 1 - (backcast_length + forecast_length), backcast_length, 2))
            time_series_cleaned_forlearning_y = np.zeros(
                (time_series.shape[0] + 1 - (backcast_length + forecast_length), forecast_length, 2))
            for j in range(backcast_length, time_series.shape[0] + 1 - forecast_length):
                time_series_cleaned_forlearning_x[j - backcast_length] = time_series[j - backcast_length:j, :]
                time_series_cleaned_forlearning_y[j - backcast_length] = time_series[j: j + forecast_length, :]

        x = np.vstack((x, time_series_cleaned_forlearning_x))
        y = np.vstack((y, time_series_cleaned_forlearning_y))

    return x, None, y

def process_data_price():
    filename = '../examples/data/nrj/EPEX_spot_DA_auction_hour_prices_20070720-20170831.csv'

    x_tl = []
    headers = True
    with open(filename, "r") as file:
        reader = csv.reader(file, delimiter=',')
        for line in reader:
            if not headers:
                x_tl.append(line)
            if headers:
                headers = False
    x_tl = [float(x_tl[i][1]) for i in range(len(x_tl)) if '00:00:00' in x_tl[i][0]]
    x_tl = np.array(x_tl)

    return x_tl

def process_data_load():
    filename = '../examples/data/nrj/20150101-20170830-forecast_load_renewable_gen.csv'

    x_tl = []
    headers = True
    with open(filename, "r") as file:
        reader = csv.reader(file, delimiter=',')
        for line in reader:
            if not headers:
                x_tl.append(line)
            if headers:
                headers = False

    x_tl = [x_tl[i][1] for i in range(len(x_tl)) if '00:00:00' in x_tl[i][0]]
    x_tl = [float(x_tl[i]) if x_tl[i] != '' else 0. for i in range(len(x_tl))]
    x_tl[x_tl == 0] = np.mean(x_tl)
    x_tl = np.array(x_tl)

    return x_tl

def process_data_gen():
    filename = '../examples/data/nrj/20150101-20170830-gen_per_prod_type.csv'

    x_tl = []
    headers = True
    with open(filename, "r") as file:
        reader = csv.reader(file, delimiter=',')
        for line in reader:
            if not headers:
                x_tl.append(line)
            if headers:
                headers = False

    x_tl = [x_tl[i][1] for i in range(len(x_tl)) if '00:00:00' in x_tl[i][0]]
    x_tl = [float(x_tl[i]) if x_tl[i] != '' else 0. for i in range(len(x_tl))]
    x_tl[x_tl == 0] = np.mean(x_tl)
    x_tl = np.array(x_tl)

    return x_tl

def get_x_y_data(backcast_length, forecast_length):
    x = np.array([]).reshape(0, backcast_length)
    y = np.array([]).reshape(0, forecast_length)

    time_series = process_data_price()[:-1]

    time_series_cleaned_forlearning_x = np.zeros(
        (time_series.shape[0] + 1 - (backcast_length + forecast_length), backcast_length))
    time_series_cleaned_forlearning_y = np.zeros(
        (time_series.shape[0] + 1 - (backcast_length + forecast_length), forecast_length))
    for j in range(backcast_length, time_series.shape[0] + 1 - forecast_length):
        time_series_cleaned_forlearning_x[j - backcast_length, :] = time_series[j - backcast_length:j]
        time_series_cleaned_forlearning_y[j - backcast_length, :] = time_series[j: j + forecast_length]
    x = np.vstack((x, time_series_cleaned_forlearning_x))
    y = np.vstack((y, time_series_cleaned_forlearning_y))

    return x.reshape((x.shape[0], x.shape[1], 1)), y.reshape((y.shape[0], y.shape[1], 1))

def get_exo_var_data(backcast_length, forecast_length):
    e1 = np.array([]).reshape(0, backcast_length)
    e2 = np.array([]).reshape(0, backcast_length)

    time_series_1 = process_data_gen()
    time_series_2 = process_data_load()
    time_series_cleaned_forlearning_1 = np.zeros(
        (time_series_1.shape[0] + 1 - (backcast_length + forecast_length), backcast_length))
    time_series_cleaned_forlearning_2 = np.zeros(
        (time_series_1.shape[0] + 1 - (backcast_length + forecast_length), backcast_length))
    for j in range(backcast_length, time_series_1.shape[0] + 1 - forecast_length):
        time_series_cleaned_forlearning_1[j - backcast_length, :] = time_series_1[j - backcast_length:j]
        time_series_cleaned_forlearning_2[j - backcast_length, :] = time_series_2[j - backcast_length:j]
    e1 = np.vstack((e1, time_series_cleaned_forlearning_1))
    e2 = np.vstack((e2, time_series_cleaned_forlearning_2))

    return e1, e2

def get_nrj_data(backcast_length, forecast_length, is_training=True):
    x, y = get_x_y_data(backcast_length, forecast_length)
    e1, e2 = get_exo_var_data(backcast_length, forecast_length)

    x_max = np.amax(np.abs(x[:90 * x.shape[0] // 100, :, :]), axis=(0, 1))
    e1_max = np.amax(np.abs(e1[:90 * x.shape[0] // 100, :]), axis=(0, 1))
    e2_max = np.amax(np.abs(e2[:90 * x.shape[0] // 100, :]), axis=(0, 1))

    x = x / x_max
    y = y / x_max
    e1 = e1 / e1_max
    e2 = e2 / e2_max

    e = np.concatenate((e1.reshape((e1.shape[0], e1.shape[1], 1)), e2.reshape((e2.shape[0], e2.shape[1], 1))), axis=-1)

    if is_training:
        return x[:90 * x.shape[0] // 100], e[:90 * x.shape[0] // 100], y[:90 * x.shape[0] // 100]
    else:
        return x[90 * x.shape[0] // 100:], e[90 * x.shape[0] // 100:], y[90 * x.shape[0] // 100:]

ext: py

max_stars_repo_name: florisrc/n-beats

max_issues_repo_name: florisrc/n-beats

max_forks_repo_name: florisrc/n-beats

reason: import numpy


text:
# -*- coding: utf-8 -*-

import numpy as np
import time

def now():
	return str(time.strftime('%Y-%m-%d %H:%M:%S'))

def save_pr(out_dir, name, epoch, pre, rec, fp_res=None, opt=None):
	if opt is None:
		out = open('{}/{}_{}_PR.txt'.format(out_dir, name, epoch + 1), 'w')
	else:
		out = out = open('{}/{}_{}_{}_PR.txt'.format(out_dir, name, opt, epoch + 1), 'w')
	if fp_res is not None:
		fp_out = open('{}/{}_{}_FP.txt'.format(out_dir, name, epoch + 1), 'w')
		for idx, r, p in fp_res:
			fp_out.write('{}{}{}\n'.format(idx, r, p))
	fp_out.close()

	for p, r in zip(pre, rec):
		out.write('{} {}\n'.format(p, r))
	out.close()

def eval_metric(true_y, pred_y, pred_p):
	'''
	calculate the precision and recall for p-r curve
	reglect the NA relation
	'''
	assert len(true_y) == len(pred_y)
	positive_num = len([i for i in true_y if i[0] > 0])
	index = np.argsort(pred_p)[::-1]

	tp = 0
	fp = 0
	fn = 0
	all_pre = [0]
	all_rec = [0]
	fp_res = []

	for idx in range(len(true_y)):
		i = true_y[idx]
		j = pred_y[idx]

		if i[0] == 0:
			if j > 0:
				fp_res.append(index[idx], j, pred_p[index[idx]])
				fp += 1
		else:
			if j == 0:
				fn += 1
			else:
				for k in i:
					if k == -1:
						break
					if k == j:
						tp += 1
						break
		if fp + tp == 0:
			precision = 1.0
		else:
			precision = tp * 1.0 / (fp + tp)
	recall = tp * 1.0 / positive_num
	if precision != all_pre[-1] or recall != all_rec[-1]:
		all_pre.append(precision)
		all_rec.append(recall)

	print("tp={}; fp={}; fn={}; positive_num={}".format(tp, fp, fn, positive_num))
	return all_pre[1:], all_rec[1:], fp_res

max_issues_repo_name: jiaolongxue/Knowledge-Graph

max_forks_repo_name: jiaolongxue/Knowledge-Graph


text:
# -*- coding: UTF-8 -*-
import sys
import cv2
import numpy as np
from matplotlib import pyplot as plt
# import matplotlib.pyplot as plt

# def cv_show(name,img):
#     cv2.imshow(name,img)
#     cv2.waitKey(0)
#     cv2.destroyAllWindows()

# cv_show('img1',img1)
# cv_show('img2',img2)
def a(imgPath1,imgPath2):
    img2 = cv2.imread(imgPath2,cv2.IMREAD_GRAYSCALE)
    img1 = cv2.imread(imgPath1,cv2.IMREAD_GRAYSCALE)

    orb = cv2.ORB_create()
    kp1,des1 = orb.detectAndCompute(img1,None)
    kp2,des2 = orb.detectAndCompute(img2,None)

    #暴力匹配
    bf = cv2.BFMatcher(cv2.NORM_HAMMING,crossCheck=True)
    matches = bf.match(des1,des2)
    matches = sorted(matches,key=lambda x:x.distance)
    img3 = cv2.drawMatches(img1,kp1,img2,kp2,matches[:40],img2,flags=2)

    cv2.imshow('test_ORB_bt',img3)
    cv2.imwrite('tests/feature/img_feature/test_ORB_bt.png',img3)

    cv2.waitKey(0)
    cv2.destroyAllWindows()

if __name__ == "__main__":
    print ('---------检测斑点------------')
    a("tests/feature/a.jpg","tests/feature/b.jpg")
    a("tests/feature/a.jpg","tests/feature/c.jpg")
    a("tests/feature/a.jpg","tests/feature/d.jpg")
    a("tests/feature/a.jpg","tests/feature/ah.jpg")


text:
"""
This file is part of the openPMD-viewer

It routes the calls to the data reader to either
the h5py data reader, or to openpmd-api.

Copyright 2020, openPMD-viewer contributors
Authors: Remi Lehe
License: 3-Clause-BSD-LBNL
"""
import numpy as np
import os
import re

available_backends = []

try:
    import openpmd_api as io
    from . import io_reader
    available_backends.append('openpmd-api')
except ImportError:
    pass

try:
    from . import h5py_reader
    available_backends.append('h5py')
except ImportError:
    pass

if len(available_backends) == 0:
    raise ImportError('No openPMD backend found.\n'
        'Please install either `h5py` or `openpmd-api`:\n'
        'e.g. with `pip install h5py` or `pip install openpmd-api`')

class DataReader( object ):
    """
    Class that performs various type of access the openPMD file.

    The methods of this class are agnostic of the actual backend package
    used in order to access the openPMD file (e.g. h5py or openpmd-api).
    The backend that is used in practice depends on which package is
    available on the current environment.
    """

    def __init__(self, backend):
        """
        Initialize the DataReader class.
        """
        self.backend = backend

        # Point to the correct reader module
        if self.backend == 'h5py':
            self.iteration_to_file = {}
        elif self.backend == 'openpmd-api':
            pass
        else:
            raise RuntimeError('Unknown backend: %s' % self.backend)

    def list_iterations(self, path_to_dir):
        """
        Return a list of the iterations that correspond to the files
        in this directory. (The correspondance between iterations and
        files is stored internally.)

        Parameter
        ---------
        path_to_dir : string
            The path to the directory where the hdf5 files are.

        Returns
        -------
        an array of integers which correspond to the iteration of each file
        (in sorted order)
        """
        if self.backend == 'h5py':
            iterations, iteration_to_file = \
                h5py_reader.list_files( path_to_dir )
            # Store dictionary of correspondence between iteration and file
            self.iteration_to_file = iteration_to_file
            if len(iterations) == 0:
                raise RuntimeError(
                    "Found no valid files in directory {0}.\n"
                    "Please check that this is the path to the openPMD files."
                    "(valid files must have the extension '.h5')"
                    .format(path_to_dir))
        elif self.backend == 'openpmd-api':
            # guess file ending from first file in directory
            first_file_name = None
            for file_name in os.listdir( path_to_dir ):
                if file_name.split(os.extsep)[-1] in io.file_extensions:
                    first_file_name = file_name
            if first_file_name is None:
                raise RuntimeError(
                    "Found no valid files in directory {0}.\n"
                    "Please check that this is the path to the openPMD files."
                    "(valid files must have one of the following extensions: {1})"
                    .format(path_to_dir, io.file_extensions))

            # match last occurance of integers and replace with %T wildcards
            # examples: data00000100.h5 diag4_00000500.h5 io12.0.bp
            #           te42st.1234.yolo.json scan7_run14_data123.h5
            file_path = re.sub(r'(\d+)(\.(?!\d).+$)', r'%T\2', first_file_name)

            self.series = io.Series(
                os.path.join( path_to_dir, file_path),
                io.Access.read_only )
            iterations = np.array( self.series.iterations )

        return iterations

    def read_openPMD_params(self, iteration, extract_parameters=True):
        """
        Extract the time and some openPMD parameters from a file

        Parameter
        ---------
        iteration: int
            The iteration at which the parameters should be extracted

        extract_parameters: bool, optional
            Whether to extract all parameters or only the time
            (Function execution is faster when extract_parameters is False)

        Returns
        -------
        A tuple with:
        - A float corresponding to the time of this iteration in SI units
        - A dictionary containing several parameters, such as the geometry, etc
         When extract_parameters is False, the second argument returned is None
        """
        if self.backend == 'h5py':
            filename = self.iteration_to_file[iteration]
            return h5py_reader.read_openPMD_params(
                    filename, iteration, extract_parameters)

        elif self.backend == 'openpmd-api':
            return io_reader.read_openPMD_params(
                    self.series, iteration, extract_parameters)

    def read_field_cartesian( self, iteration, field, coord, axis_labels,
                          slice_relative_position, slice_across ):
        """
        Extract a given field from an openPMD file in the openPMD format,
        when the geometry is cartesian (1d, 2d or 3d).

        Parameters
        ----------
        iteration : int
           The iteration at which to extract the fields

        field : string, optional
           Which field to extract

        coord : string, optional
           Which component of the field to extract

        axis_labels: list of strings
           The name of the dimensions of the array (e.g. ['x', 'y', 'z'])

        slice_across : list of str or None
           Direction(s) across which the data should be sliced
           Elements can be:
             - 1d: 'z'
             - 2d: 'x' and/or 'z'
             - 3d: 'x' and/or 'y' and/or 'z'
           Returned array is reduced by 1 dimension per slicing.

        slice_relative_position : list of float or None
           Number(s) between -1 and 1 that indicate where to slice the data,
           along the directions in `slice_across`
           -1 : lower edge of the simulation box
           0 : middle of the simulation box
           1 : upper edge of the simulation box

        Returns
        -------
        A tuple with
           F : a ndarray containing the required field
           info : a FieldMetaInformation object
           (contains information about the grid; see the corresponding docstring)
        """
        if self.backend == 'h5py':
            filename = self.iteration_to_file[iteration]
            return h5py_reader.read_field_cartesian(
                filename, iteration, field, coord, axis_labels,
                slice_relative_position, slice_across )
        elif self.backend == 'openpmd-api':
            return io_reader.read_field_cartesian(
                self.series, iteration, field, coord, axis_labels,
                slice_relative_position, slice_across )

    def read_field_circ( self, iteration, field, coord, slice_relative_position,
                        slice_across, m=0, theta=0. ):
        """
        Extract a given field from an openPMD file in the openPMD format,
        when the geometry is thetaMode

        Parameters
        ----------
        iteration : int
           The iteration at which to extract the fields

        field : string, optional
           Which field to extract
           Either 'rho', 'E', 'B' or 'J'

        coord : string, optional
           Which component of the field to extract
           Either 'r', 't' or 'z'

        m : int or string, optional
           The azimuthal mode to be extracted

        theta : float or None
           Angle of the plane of observation with respect to the x axis
           If `theta` is not None, then this function returns a 2D array
           corresponding to the plane of observation given by `theta` ;
           otherwise it returns a full 3D Cartesian array

        slice_across : list of str or None
           Direction(s) across which the data should be sliced
           Elements can be 'r' and/or 'z'
           Returned array is reduced by 1 dimension per slicing.

        slice_relative_position : list of float or None
           Number(s) between -1 and 1 that indicate where to slice the data,
           along the directions in `slice_across`
           -1 : lower edge of the simulation box
           0 : middle of the simulation box
           1 : upper edge of the simulation box

        Returns
        -------
        A tuple with
           F : a 3darray or 2darray containing the required field,
               depending on whether `theta` is None or not
           info : a FieldMetaInformation object
           (contains information about the grid; see the corresponding docstring)
        """
        if self.backend == 'h5py':
            filename = self.iteration_to_file[iteration]
            return h5py_reader.read_field_circ(
                filename, iteration, field, coord, slice_relative_position,
                slice_across, m, theta )
        elif self.backend == 'openpmd-api':
            return io_reader.read_field_circ(
                self.series, iteration, field, coord, slice_relative_position,
                slice_across, m, theta )

    def read_species_data( self, iteration, species, record_comp, extensions):
        """
        Extract a given species' record_comp

        Parameters
        ----------
        iteration: int
            The iteration at which to extract the species data

        species: string
            The name of the species to extract (in the openPMD file)

        record_comp: string
            The record component to extract
            Either 'x', 'y', 'z', 'ux', 'uy', 'uz', or 'w'

        extensions: list of strings
            The extensions that the current OpenPMDTimeSeries complies with
        """
        if self.backend == 'h5py':
            filename = self.iteration_to_file[iteration]
            return h5py_reader.read_species_data(
                    filename, iteration, species, record_comp, extensions )
        elif self.backend == 'openpmd-api':
            return io_reader.read_species_data(
                    self.series, iteration, species, record_comp, extensions )

    def get_grid_parameters(self, iteration, avail_fields, metadata ):
        """
        Return the parameters of the spatial grid (grid size and grid range)
        in two dictionaries

        Parameters:
        -----------
        iteration: int
            The iteration at which to extract the parameters

        avail_fields: list
           A list of the available fields
           e.g. ['B', 'E', 'rho']

        metadata: dictionary
          A dictionary whose keys are the fields of `avail_fields` and
          whose values are dictionaries that contain metadata (e.g. geometry)

        Returns:
        --------
        A tuple with `grid_size_dict` and `grid_range_dict`
        Both objects are dictionaries, with their keys being the labels of
        the axis of the grid (e.g. 'x', 'y', 'z')
        The values of `grid_size_dict` are the number of gridpoints along
        each axis.
        The values of `grid_range_dict` are lists of two floats, which
        correspond to the min and max of the grid, along each axis.
        """
        if self.backend == 'h5py':
            filename = self.iteration_to_file[iteration]
            return h5py_reader.get_grid_parameters(
                filename, iteration, avail_fields, metadata )
        elif self.backend == 'openpmd-api':
            return io_reader.get_grid_parameters(
                self.series, iteration, avail_fields, metadata )

max_stars_repo_name: AngelFP/openPMD-viewer

max_issues_repo_name: AngelFP/openPMD-viewer

max_forks_repo_name: AngelFP/openPMD-viewer