text:
"""
By Peng Wu, 04/08/2020
Data processing module, based on the Matlab implementation by Enrico Anderlini.
This script:

0. Automatic load all raw data of multiple vehicles with multiple parameters

1. Filters:
    a. Keep only the dive profiles for which there are data from all sensors:
        i. Manually set the sensors not of interests or automatically remove parameters with high empty rates
        ii. Ignore data with empty record of interested measurements
    b. Remove dives where the depth did not reach a depth of 25 m
        i. 25m filtering of depth
        ii. Minimum number of measurement points filtering

2. Resampling
    a. Apply uniformed timeline for all measurements by 1D interpolation

3. Processing for data-driven analysis
    a. Normalisation

# Note:
-   the dictionary 'data' is always the data collection currently under construction in all processing procedures
-   in all processing functions, the input data is not changed, we simply use input data to construct the processed data
-   the data is structured as nested dictionaries:
    data[vehicle][cycle][parameter][records]
    where:
        vehicle is the vehicle name
        cycle is the cycle number
        parameter is the name of the sensor/parameter
        records is the actual measurements
"""

import numpy as np
import os
import pickle
from scipy import interpolate
import time
from copy import deepcopy
from slocum import Slocum
import pandas as pd
import random

class DataProcessingNew:
    def __init__(self, verbose=True, auto_remove=False):
        """
        Locate the data path
        Set function parameters
        Initialise intermediate data
        """
        # set numpy random seed
        np.random.seed(111)

        # Raw data
        self.data_raw = {}

        # filtered data
        self.data_filter = {}

        # interpolated data
        self.data_interp = {}

        # resampled data
        self.data_resample = {}

        # data for model based
        self.data_model = {}

        # data for data-driven
        self.data_deep = {}

        self.path = '/media/peng/Data/PengData/SlocumData/Postprocessed'

        self.verbose = verbose

        # function timer
        self.timer = None
        # time step for interpolation, default 5 s
        self.dt = 5

        # minimum no. of points
        self.np = 9

        # empty rate threshold
        self.e_r_threshold = 0.53

        # TODO: user defined threshold, this should be defined by the user after initial processing
        # currently only 'depth < 25' is applied, also need set < > or = etc.
        self.sensor_threshold = {'depth': 25}

        # max and min values of parameters
        self.minmax = None
        self.minmax_multi = None

        # automatically or manually remove parameters not of interests
        self.auto_remove = auto_remove

        # set parameters not of interests if auto_remove is disabled
        # TODO: this is to be chosen by the user through tick boxes if there is a GUI
        self.not_interested = ['heading_control',
                               'pitch_control',
                               'rudder_angle_control',
                               'vbd_control']

        # store maximum time for each dive
        self.ft = None

        # store raw time lines for each vehicle
        self.t = None

        # automatically detect vehicle names and parameters
        self.sensors = {}
        names = os.listdir(self.path)
        for name in names:
            if os.path.isdir(os.path.join(self.path, name)):
                dives = os.listdir(os.path.join(self.path, name))
                for i in dives:
                    if os.path.isdir(os.path.join(self.path, name, i)):
                        temp = os.listdir(os.path.join(self.path, name, i))
                        for sensor in temp:
                            base = os.path.basename(os.path.join(self.path, name, sensor))
                            base = os.path.splitext(base)[0]
                            self.sensors[base] = {}
                        break
                    else:
                        pass
                break

        # print the parameters in the raw data
        if self.verbose:
            print('User defined raw data path is', self.path)

            print('The vehicles are', names)

            print('The measurements are:')
            for key, value in self.sensors.items():
                print(f'{key}: {value}')
            print('Initialisation done.')

    def load_raw(self):
        """
        Read the raw data in *.out files
        And output the data stored in Python dictionary
        """

        # the dictionary to store raw data
        data_raw = {}
        print('Extracting raw data...')
        self.timer = time.time()
        # get the folders' names relating to each glider:
        names = os.listdir(self.path)
        vehicles = {}

        for name in names:
            if os.path.isdir(os.path.join(self.path, name)):
                vehicles[name] = {}

        for vehicle in vehicles:

            if self.verbose:
                print(vehicle)

            cycle_profiles = {}

            current_dir = os.path.join(self.path, vehicle)
            dirlist = []

            # get the list of all cycles
            cycle_list = os.listdir(current_dir)
            for cycle_sequence in cycle_list:
                if os.path.isdir(os.path.join(current_dir, cycle_sequence)):
                    dirlist.append(cycle_sequence)

            # initialise data, see header about data
            data = {}

            for cycle in dirlist:
                # initialise list of sensors
                sensors = deepcopy(self.sensors)
                for sensor in sensors:
                    data_path = os.path.join(self.path, vehicle, cycle, sensor + '.out')
                    # read the records if it is not empty, otherwise this sensor/parameter is emtpy
                    size = os.stat(data_path).st_size
                    if size:
                        sensors[sensor] = np.genfromtxt(data_path)
                    else:
                        sensors[sensor] = np.array([])

                cycle_profiles[cycle] = deepcopy(sensors)

            data[vehicle] = deepcopy(cycle_profiles)

            # create data folder if there is no
            if not os.path.exists('data'):
                os.makedirs('data')

            # save extracted data to 'data' folder
            pickle.dump(cycle_profiles, open(os.path.join('data', 'rawdata_' + vehicle + '.pkl'), 'wb'))
            if self.verbose:
                print(vehicle, 'raw data loaded & saved to pickle file')

        self.data_raw = deepcopy(data)

        if self.verbose:
            print("--- %f seconds ---" % (time.time() - self.timer))

        return data

    def filter(self):
        """
        This function filters:
        Objective a:
            i. Manually set the sensors not of interests or automatically remove parameters with high empty rates
                -   parameters with high empty rates
                -   parameters not of interests defined by user
            ii. Ignore data with empty record of interested measurements
        Objective b:
            i. 25m filtering of depth,
                TODO: this should be a generic user defined parameter, including the sensors, and their limits
            ii. Minimum number of measurement points filtering
        :return: filtered cycles
        """
        print('Filtering in process...')
        self.timer = time.time()

        if self.auto_remove:
            removed_sensors = self.find_empty_rate()
        else:
            removed_sensors = self.not_interested

        # load raw data
        data_raw = deepcopy(self.data_raw)

        # initialise target data
        data = {}

        # clear memory
        del self.data_raw

        if self.verbose:
            print('Parameters to be removed are:', removed_sensors)

        # remove sensors not of interests
        sensors_filtered = deepcopy(self.sensors)
        for sensor in removed_sensors:
            sensors_filtered.pop(sensor, None)

        for vehicle, cycles in data_raw.items():
            data[vehicle] = {}
            for cycle, sensors in cycles.items():
                data[vehicle][cycle] = {}
                for sensor in sensors_filtered:
                    if sensor in data_raw[vehicle][cycle]:
                        data[vehicle][cycle][sensor] = deepcopy(data_raw[vehicle][cycle][sensor])

        # remove cycles if they contain empty parameters of interest, or the record of any parameter is less than np
        empty = {}
        for vehicle, cycles in data.items():
            empty[vehicle] = []
            for cycle, sensors in cycles.items():
                for sensor, records in sensors.items():
                    if records.shape[0] <= self.np:
                        empty[vehicle].append(cycle)
                        break

        for vehicle in data:
            for cycle in empty[vehicle]:
                del data[vehicle][cycle]
                # print(vehicle, cycle)

        # delete cycles with parameters beyond defined thresholds
        thres = {}
        for vehicle, cycles in data.items():
            thres[vehicle] = []
            for cycle, sensors in cycles.items():
                for r, l in self.sensor_threshold.items():
                    if np.max(sensors[r][:, 1]) < l:
                        thres[vehicle].append(cycle)

        for vehicle in data:
            for cycle in thres[vehicle]:
                del data[vehicle][cycle]

        # remove vehicle if it is empty
        empty_vehicle = []
        for vehicle, cycles in data.items():
            if len(cycles) == 0:
                empty_vehicle.append(vehicle)

        for vehicle in empty_vehicle:
            del data[vehicle]
            if self.verbose:
                print(vehicle, 'is removed due to lack of valuable data: '
                               'try to relax the filtering limits if this vehicle is of interest!')

        """
        to get the raw timelines, including starting and ending time (in raw format),
        this is to ensure cycles in sequence
        """
        t_pd = {}
        for vehicle, cycles in data.items():
            temp = []
            for cycle, sensors in cycles.items():
                st_raw = np.array([])
                ft_raw = np.array([])
                for sensor, records in sensors.items():
                    st_raw = np.append(st_raw, records[0, 0])
                    ft_raw = np.append(ft_raw, records[-1, 0])

                temp.append(
                    {'cycle': cycle, 's_t': np.min(st_raw), 'f_t': np.max(ft_raw)}
                )

            t_pd[vehicle] = pd.DataFrame(temp)

        t_surface = {}

        for vehicle, temp in t_pd.items():
            t_surface[vehicle] = {}
            temp.sort_values('s_t', inplace=True, ascending=True, ignore_index=False)
            length = temp.shape[0]
            for i in range(length):
                if i == (length-1):
                    t_surface[vehicle][temp.iloc[i, 0]] = 0
                else:
                    t_surface[vehicle][temp.iloc[i, 0]] = temp.iloc[i + 1, 1] - temp.iloc[i, 2]
        self.t = t_surface

        # get vertical velocity

        data = self.get_velocity(data)

        self.data_filter = deepcopy(data)

        if self.verbose:
            print("--- %f seconds ---" % (time.time() - self.timer))

        return data

    def interpolate(self):

        """
        Apply uniformed timeline for all measurements by 1D interpolation
        -   Start time
        -   End time
        -   Relative timelines
        -   Interpolations
        """
        print('Interpolation in process...')
        self.timer = time.time()

        data = deepcopy(self.data_filter)

        # set relative timelines for dive profiles with complete records
        ft = {}
        for vehicle, cycles in data.items():
            ft[vehicle] = {}
            for cycle in cycles:
                start_time = np.array([])
                final_time = np.array([])
                # get start and final times of all measurements, find min and max of all measurements for each cycle
                for sensor, measurement in data[vehicle][cycle].items():
                    start_time = np.append(start_time, measurement[0, 0])
                    final_time = np.append(final_time, measurement[-1, 0])

                s_t = np.min(start_time)

                # set relative start time
                for sensor in cycles[cycle]:
                    data[vehicle][cycle][sensor][:, 0] -= s_t

                ft[vehicle][cycle] = np.max(final_time) - s_t

                # apply 1d interpolation to all measurements
                t = np.arange(0, ft[vehicle][cycle], self.dt)

                for sensor, records in cycles[cycle].items():
                    f = interpolate.interp1d(records[:, 0], records[:, 1], fill_value='extrapolate')
                    y = f(t)
                    # clear original data
                    data[vehicle][cycle][sensor] = np.zeros([len(t), 2])
                    # fill in with interpolated data
                    data[vehicle][cycle][sensor][:, 0] = deepcopy(t)
                    data[vehicle][cycle][sensor][:, 1] = deepcopy(y)

        self.ft = ft

        self.data_interp = data

        if self.verbose:
            print("--- %f seconds ---" % (time.time() - self.timer))

        return data

    def model_based_processing(self):

        """
        Processing for model-based analysis:
            a. New parameters introduced and calculation required (Fourth pass, gsw_toolbox called)
            b. Divide the cycles into dives and climbs and remove transients (Fifth pass)
            c. Put together the dives and climbs without transients (Sixth pass or Seventh pass)
            d. Compute the bouncy buoyancy value for each data point (Eighth pass)
        :return: data for system identification
        """
        data_interp = deepcopy(self.data_interp)

        s = Slocum(self.ft, self.dt)

        data_calculated = s.slocum_cal_a(data_interp)

        dive, climb, data = s.slocum_cal_b(data_calculated, self.t)

        data_new = s.slocum_cal_c(dive, climb, data)

        # data for system identification
        data_b = s.slocum_cal_d(data_new)

        return data_b

    def data_driven_processing(self):
        """
        This function process the data for data driven approaches
        TODO: currently only normalisation is included
        :return: normalised data
        """
        print('Prepare data for data-driven approaches...')
        self.timer = time.time()
        data = deepcopy(self.data_interp)

        # normalise the data
        data_normalised_multi = self.normalise_multi(deepcopy(data))

        self.data_multi_train(data_normalised_multi)

        if self.verbose:
            print("--- %f seconds ---" % (time.time() - self.timer))

    def load_extracted(self):
        """
        Load previously extracted
        """
        print('Load pickle files...')
        self.timer = time.time()

        files = os.listdir('data')
        vehicles = []
        data = {}
        for file in files:
            base = os.path.basename(file)
            vehicle = os.path.splitext(base)[0][8:]
            vehicles.append(vehicle)    # remove 'raw_data_' string
            data[vehicle] = pickle.load(open(os.path.join('data', file), 'rb'))
            print(vehicle)

        self.data_raw = deepcopy(data)

        if self.verbose:
            print("--- %f seconds ---" % (time.time() - self.timer))

        return data

    def get_velocity(self, data):
        # replace depth to vertical_velocity for calculation
        for vehicle, cycles in data.items():
            for cycle in cycles:
                data[vehicle][cycle]['vertical_velocity'] = data[vehicle][cycle]['depth'][0:-1, :]

                data[vehicle][cycle]['vertical_velocity'][:, 1] = \
                    np.diff(data[vehicle][cycle]['depth'][:, 1])/np.diff(data[vehicle][cycle]['depth'][:, 0])
        return data

    def find_empty_rate(self):
        """
        This function finds the empty rates of sensors/parameters

        """
        empty_rate = deepcopy(self.sensors)

        for sensor in empty_rate:
            empty_rate[sensor] = 0.

        total_cycle = 0

        data = self.data_raw
        for vehicle, cycles in data.items():
            for cycle, sensors in cycles.items():
                total_cycle += 1
                for sensor in sensors:
                    size = len(sensors[sensor])
                    if size == 0:
                        empty_rate[sensor] += 1
        for sensor in empty_rate:
            empty_rate[sensor] /= total_cycle

        empty_rate_sorted = sorted(empty_rate.items(), key=lambda x: x[1], reverse=True)

        empty_high = []

        if self.verbose and empty_rate_sorted[0][1] > self.e_r_threshold:
            print('Be careful: the following sensors/parameters have empty rates higher than the threshold of',
                  self.e_r_threshold, '!!!')

        for i in empty_rate_sorted:
            if i[1] > self.e_r_threshold:
                empty_high.append(i[0])
                if self.verbose:
                    print(i[0], ':', i[1])

        print(empty_rate_sorted)
        print('Total cycle number is', total_cycle)

        return empty_high

    def normalise(self, data):
        """
        Normalise all data to the range of [0, 1]
        """
        minmax = {}

        # initialise the normalised data
        for vehicle, cycles in data.items():
            minmax[vehicle] = {}
            for cycle, sensors in cycles.items():
                for sensor in sensors:
                    minmax[vehicle][sensor] = {}
                    minmax[vehicle][sensor]['min'] = None
                    minmax[vehicle][sensor]['max'] = None

        # find min and max for sensors
        for vehicle, cycles in data.items():
            for cycle, sensors in cycles.items():
                for sensor, records in sensors.items():
                    r = records[:, 1]
                    min_r = r.min()
                    max_r = r.max()
                    if minmax[vehicle][sensor]['min'] is None:
                        minmax[vehicle][sensor]['min'] = min_r
                    else:
                        minmax[vehicle][sensor]['min'] = min(min_r, minmax[vehicle][sensor]['min'])

                    if minmax[vehicle][sensor]['max'] is None:
                        minmax[vehicle][sensor]['max'] = max_r
                    else:
                        minmax[vehicle][sensor]['max'] = max(max_r, minmax[vehicle][sensor]['max'])

        # normalise sensor measurements to [0, 1]
        for vehicle, cycles in data.items():
            for cycle, sensors in cycles.items():
                for sensor, records in sensors.items():
                    ptp = minmax[vehicle][sensor]['max'] - minmax[vehicle][sensor]['min']
                    data[vehicle][cycle][sensor][:, 1] = (records[:, 1] - minmax[vehicle][sensor]['min']) / ptp
                    # data[vehicle][cycle][sensor][:, 1] = 2 * (records[:, 1] - minmax[vehicle][sensor]['min'])/ptp - 1
        self.minmax = minmax

        # save the minimum and maximum values to file
        if not os.path.exists('training_data'):
            os.makedirs('training_data')

        pickle.dump(minmax, open(os.path.join('training_data', 'de_normal.pkl'), 'wb'))
        return data

    def data_multi_train(self, data):
        """
        This function prepares the data of multiple vehicles of the same type
        -   Training data structure:
            -   Patch: small patches from the processed cycle data
                -   User to define which sensors/parameters not to include
                    -   Currently ignore the location data to ensure transferability
            -   Vehicle: vehicle deployment
            -   Cycle: cycle number/sequence

        -   Generates the:
            -   Training dataset (can also be extended test dataset)
            -   Test dataset (unseen by the model)
            -   Sensitivity study dataset
        """
        no_patches = int(1e5)
        no_steps = 64
        no_patches_test = 10
        # vehicles to be included are as below, N.B. the three Slocum deployments are healthy
        vehicle_list = ['unit_345_d1', 'unit_397_d1']
        print('Deployments for training are:', vehicle_list)
        # remove sensors not to include
        sensor_list = []
        not_to_include = ['gps_lat', 'gps_lon', 'lat', 'lon', 'depth']

        # extract the first XX cycles data
        training_cycles = 400

        # remove vehicles to be used for test
        not_vehicle = []
        for vehicle, cycles in data.items():
            print(len(cycles))
            if vehicle not in vehicle_list:
                not_vehicle.append(vehicle)
                print('Vehicle', vehicle, 'is not included due to insufficient cycles, only', len(cycles), 'cycles')
        print('Deployments for test are:', not_vehicle)

        for vehicle in vehicle_list:
            cycles = data[vehicle]
            for cycle, sensors in cycles.items():
                for sensor in sensors:
                    sensor_list.append(sensor)
                break
            break

        patches_per_vehicle = int(no_patches/len(vehicle_list))
        patches_per_cycle = int(patches_per_vehicle / training_cycles)

        for sensor in not_to_include:
            sensor_list.remove(sensor)

        # create data folder if there is no
        if not os.path.exists('training_data'):
            os.makedirs('training_data')

        # create data folder if there is no
        if not os.path.exists('test_data'):
            os.makedirs('test_data')

        # save sensor list to to training_data folder
        pickle.dump(sensor_list, open(os.path.join('training_data', 'sensor_list_multi' + '.pkl'), 'wb'))
        if self.verbose:
            print('Sensor list for multi vessels saved to pickle file in training_data folder')

        # process multi-vehicle training data

        dataset = {'data': np.zeros([no_patches, len(sensor_list), no_steps]),
                   'vehicle': [None] * no_patches,
                   'cycle': np.zeros([no_patches])}

        c = 0
        for vehicle in vehicle_list:
            cycles = data[vehicle]
            print('Processing training data of', vehicle, len(cycles), 'patches:', patches_per_cycle)

            cycle_list = list(cycles)

            for _ in range(training_cycles):
                cycle = random.choice(cycle_list)
                sensors = cycles[cycle]
                for k in range(patches_per_cycle):
                    patch = np.empty([len(sensor_list), no_steps])
                    start_idx = None
                    for i in range(len(sensor_list)):
                        sensor = sensor_list[i]
                        if start_idx is None:
                            start_idx = np.random.randint(low=0, high=sensors[sensor].shape[0] - no_steps)
                            end_idx = start_idx + no_steps
                        patch[i, :] = sensors[sensor][:, 1][start_idx:end_idx]

                    if c < no_patches:
                        dataset['data'][c, :, :] = patch
                        dataset['cycle'][c] = cycle
                        dataset['vehicle'][c] = vehicle
                    else:
                        break
                    c += 1

        # save the training data to hard drive
        pickle.dump(dataset, open(os.path.join('training_data', 'multi_vehicle.pkl'), 'wb'))
        if self.verbose:
            print('Training data of multi_vehicle saved to pickle file in training_data folder')

        # prepare test data, note this part is modified for sensitivity study
        dts = np.array([5, 10, 30, 60, 120, 240])    # change here to modify the sensitivity study settings

        # remove 'unit_419_d1' as it is the healthy reference
        not_vehicle.remove('unit_399_d2')
        not_vehicle.remove('unit_205')
        # resample, then interpolate, every time work on one sensor and keep other sensors unchanged
        for s in sensor_list:
            for dt in dts:
                data_val = deepcopy(data)
                for vehicle in not_vehicle:
                    cycles = data_val[vehicle]
                    for cycle, sensors in cycles.items():
                        # restore the interpolated data
                        records = data[vehicle][cycle][s]
                        x = records[:, 0]
                        y = records[:, 1]
                        # resample
                        new_x = np.arange(0, records[-1, 0], dt)
                        f = interpolate.interp1d(x, y, fill_value='extrapolate')
                        new_y = f(new_x)
                        ff = interpolate.interp1d(new_x, new_y, fill_value='extrapolate')

                        # interpolate
                        time_line = np.arange(0, self.ft[vehicle][cycle], 5)
                        data_val[vehicle][cycle][s][:, 0] = deepcopy(time_line)
                        data_val[vehicle][cycle][s][:, 1] = ff(time_line)

                test_data_all = {}
                sensitivity_idx = pickle.load(open(os.path.join('test_data', 'sensitivity_idx.pkl'), 'rb'))
                for vehicle in not_vehicle:
                    print('Test data of', vehicle)
                    cycles = data_val[vehicle]
                    test_cycles = len(cycles)
                    no_patches = test_cycles * no_patches_test

                    test_data = {'data': np.zeros([no_patches, len(sensor_list), no_steps]),
                                       'vehicle': [None] * no_patches,
                                       'cycle': np.zeros([no_patches])}
                    c = 0
                    for cycle, sensors in cycles.items():
                        patch = np.empty([len(sensor_list), no_steps])
                        start_idx_list = sensitivity_idx[vehicle][cycle]
                        for i in range(no_patches_test):
                            start_idx = start_idx_list[i]
                            end_idx = start_idx + no_steps
                            for k in range(len(sensor_list)):
                                sensor = sensor_list[k]
                                patch[k, :] = sensors[sensor][:, 1][start_idx:end_idx]
                            if c < no_patches:
                                test_data['data'][c, :, :] = patch
                                test_data['cycle'][c] = cycle
                                test_data['vehicle'][c] = vehicle
                            else:
                                break
                            c += 1

                    test_data_all[vehicle] = deepcopy(test_data)
                # save test data to 'test_data' folder
                file_name = 'test_data_all_' + s + '_' + str(dt) + '.pkl'
                pickle.dump(test_data_all, open(os.path.join('test_data', file_name), 'wb'))
                if self.verbose:
                    print('Test data of multi_vehicle saved to pickle file in training_data folder', file_name)

        test_data_group_b_ = {}
        for dt in dts:
            data_val = deepcopy(data)
            for vehicle in not_vehicle:
                cycles = data_val[vehicle]
                for cycle, sensors in cycles.items():
                    for s, records in sensors.items():
                        # restore the interpolated data
                        records = data[vehicle][cycle][s]
                        x = records[:, 0]
                        y = records[:, 1]
                        # resample
                        new_x = np.arange(0, records[-1, 0], dt)
                        f = interpolate.interp1d(x, y, fill_value='extrapolate')
                        new_y = f(new_x)
                        ff = interpolate.interp1d(new_x, new_y, fill_value='extrapolate')
                        # interpolate
                        time_line = np.arange(0, self.ft[vehicle][cycle], 5)
                        data_val[vehicle][cycle][s][:, 0] = deepcopy(time_line)
                        data_val[vehicle][cycle][s][:, 1] = ff(time_line)
                test_data = {'data': np.zeros([no_patches, len(sensor_list), no_steps]),
                             'vehicle': [None] * no_patches,
                             'cycle': np.zeros([no_patches])}
                c = 0
                for cycle, sensors in cycles.items():
                    patch = np.empty([len(sensor_list), no_steps])
                    start_idx_list = sensitivity_idx[vehicle][cycle]
                    for i in range(no_patches_test):
                        start_idx = start_idx_list[i]
                        end_idx = start_idx + no_steps
                        for k in range(len(sensor_list)):
                            sensor = sensor_list[k]
                            patch[k, :] = sensors[sensor][:, 1][start_idx:end_idx]
                        if c < no_patches:
                            test_data['data'][c, :, :] = patch
                            test_data['cycle'][c] = cycle
                            test_data['vehicle'][c] = vehicle
                        else:
                            break
                        c += 1
                test_data_group_b_[vehicle] = deepcopy(test_data)
            file_name = 'test_data_group_b_' + str(dt) + '.pkl'
            pickle.dump(test_data_group_b_, open(os.path.join('test_data', file_name), 'wb'))
            if self.verbose:
                print('Test data of multi_vehicle saved to pickle file in training_data folder', file_name)

    def normalise_multi(self, data):
        """
        This normalise the dat for all vehicles to multi-vehicle training, all data are to the range of [0, 1]
        """
        minmax = {}

        # initialise the normalised data
        sensor_list = []
        for vehicle, cycles in data.items():
            for cycle, sensors in cycles.items():
                for sensor in sensors:
                    sensor_list.append(sensor)
                break
            break

        for sensor in sensor_list:
            minmax[sensor] = {}
            minmax[sensor]['min'] = None
            minmax[sensor]['max'] = None

        # find min and max for the sensors across the vehicles
        for vehicle, cycles in data.items():
            for cycle, sensors in cycles.items():
                for sensor, records in sensors.items():
                    r = records[:, 1]
                    min_r = r.min()
                    max_r = r.max()
                    if minmax[sensor]['min'] is None:
                        minmax[sensor]['min'] = min_r
                    else:
                        minmax[sensor]['min'] = min(min_r, minmax[sensor]['min'])

                    if minmax[sensor]['max'] is None:
                        minmax[sensor]['max'] = max_r
                    else:
                        minmax[sensor]['max'] = max(max_r, minmax[sensor]['max'])

        # normalise sensor measurements to [0, 1]
        for vehicle, cycles in data.items():
            for cycle, sensors in cycles.items():
                for sensor, records in sensors.items():
                    ptp = minmax[sensor]['max'] - minmax[sensor]['min']
                    data[vehicle][cycle][sensor][:, 1] = (records[:, 1] - minmax[sensor]['min']) / ptp
                    # data[cycle][sensor][:, 1] = 2 * (records[:, 1] - minmax[sensor]['min'])/ptp - 1
        self.minmax_multi = minmax

        # save the minimum and maximum values to file
        if not os.path.exists('training_data'):
            os.makedirs('training_data')

        pickle.dump(minmax, open(os.path.join('training_data', 'de_normal_multi.pkl'), 'wb'))
        return data

if __name__ == "__main__":

    dp = DataProcessingNew()
    _ = dp.load_extracted()
    _ = dp.filter()
    _ = dp.interpolate()

    dp.data_driven_processing()

reason: import numpy,from scipy


text:
#!/usr/bin/python

import numpy
import sys
import operator

from frc971.control_loops.python import control_loop
from frc971.control_loops.python import controls

from y2016.control_loops.python.shoulder import Shoulder, IntegralShoulder
from y2016.control_loops.python.wrist import Wrist, IntegralWrist
from aos.util.trapezoid_profile import TrapezoidProfile

from matplotlib import pylab
import gflags
import glog

FLAGS = gflags.FLAGS

try:
    gflags.DEFINE_bool('plot', False, 'If true, plot the loop response.')
except gflags.DuplicateFlagError:
    pass

class Arm(control_loop.ControlLoop):

    def __init__(self, name="Arm", J=None):
        super(Arm, self).__init__(name=name)
        self._shoulder = Shoulder(name=name, J=J)
        self._shooter = Wrist(name=name)
        self.shoulder_Kv = self._shoulder.Kv / self._shoulder.G

        # Do a coordinate transformation.
        # X_shooter_grounded = X_shooter + X_shoulder
        # dX_shooter_grounded/dt = A_shooter * X_shooter + A_shoulder * X_shoulder +
        #                          B_shoulder * U_shoulder + B_shooter * U_shooter
        # dX_shooter_grounded/dt = A_shooter * (X_shooter_grounded - X_shoulder) +
        #                          A_shoulder * X_shoulder + B_shooter * U_shooter + B_shoulder * U_shoulder
        # X = [X_shoulder; X_shooter + X_shoulder]
        # dX/dt = [A_shoulder                       0] [X_shoulder        ] + [B_shoulder         0] [U_shoulder]
        #         [(A_shoulder - A_shooter) A_shooter] [X_shooter_grounded] + [B_shoulder B_shooter] [ U_shooter]
        # Y_shooter_grounded = Y_shooter + Y_shoulder

        self.A_continuous = numpy.matrix(numpy.zeros((4, 4)))
        self.A_continuous[0:2, 0:2] = self._shoulder.A_continuous
        self.A_continuous[2:4, 0:2] = (
            self._shoulder.A_continuous - self._shooter.A_continuous)
        self.A_continuous[2:4, 2:4] = self._shooter.A_continuous

        self.B_continuous = numpy.matrix(numpy.zeros((4, 2)))
        self.B_continuous[0:2, 0:1] = self._shoulder.B_continuous
        self.B_continuous[2:4, 1:2] = self._shooter.B_continuous
        self.B_continuous[2:4, 0:1] = self._shoulder.B_continuous

        self.C = numpy.matrix(numpy.zeros((2, 4)))
        self.C[0:1, 0:2] = self._shoulder.C
        self.C[1:2, 0:2] = -self._shoulder.C
        self.C[1:2, 2:4] = self._shooter.C

        # D is 0 for all our loops.
        self.D = numpy.matrix(numpy.zeros((2, 2)))

        self.dt = 0.005

        self.A, self.B = self.ContinuousToDiscrete(self.A_continuous,
                                                   self.B_continuous, self.dt)

        # Cost of error
        self.Q = numpy.matrix(numpy.zeros((4, 4)))
        q_pos_shoulder = 0.014
        q_vel_shoulder = 4.00
        q_pos_shooter = 0.014
        q_vel_shooter = 4.00
        self.Q[0, 0] = 1.0 / q_pos_shoulder**2.0
        self.Q[1, 1] = 1.0 / q_vel_shoulder**2.0
        self.Q[2, 2] = 1.0 / q_pos_shooter**2.0
        self.Q[3, 3] = 1.0 / q_vel_shooter**2.0

        self.Qff = numpy.matrix(numpy.zeros((4, 4)))
        qff_pos_shoulder = 0.005
        qff_vel_shoulder = 1.00
        qff_pos_shooter = 0.005
        qff_vel_shooter = 1.00
        self.Qff[0, 0] = 1.0 / qff_pos_shoulder**2.0
        self.Qff[1, 1] = 1.0 / qff_vel_shoulder**2.0
        self.Qff[2, 2] = 1.0 / qff_pos_shooter**2.0
        self.Qff[3, 3] = 1.0 / qff_vel_shooter**2.0

        # Cost of control effort
        self.R = numpy.matrix(numpy.zeros((2, 2)))
        r_voltage = 1.0 / 12.0
        self.R[0, 0] = r_voltage**2.0
        self.R[1, 1] = r_voltage**2.0

        self.Kff = controls.TwoStateFeedForwards(self.B, self.Qff)

        glog.debug('Shoulder K')
        glog.debug(repr(self._shoulder.K))
        glog.debug(
            'Poles are %s',
            repr(
                numpy.linalg.eig(self._shoulder.A -
                                 self._shoulder.B * self._shoulder.K)[0]))

        # Compute controller gains.
        # self.K = controls.dlqr(self.A, self.B, self.Q, self.R)
        self.K = numpy.matrix(numpy.zeros((2, 4)))
        self.K[0:1, 0:2] = self._shoulder.K
        self.K[1:2, 0:2] = (
            -self.Kff[1:2, 2:4] * self.B[2:4, 0:1] * self._shoulder.K +
            self.Kff[1:2, 2:4] * self.A[2:4, 0:2])
        self.K[1:2, 2:4] = self._shooter.K

        glog.debug('Arm controller %s', repr(self.K))

        # Cost of error
        self.Q = numpy.matrix(numpy.zeros((4, 4)))
        q_pos_shoulder = 0.05
        q_vel_shoulder = 2.65
        q_pos_shooter = 0.05
        q_vel_shooter = 2.65
        self.Q[0, 0] = q_pos_shoulder**2.0
        self.Q[1, 1] = q_vel_shoulder**2.0
        self.Q[2, 2] = q_pos_shooter**2.0
        self.Q[3, 3] = q_vel_shooter**2.0

        # Cost of control effort
        self.R = numpy.matrix(numpy.zeros((2, 2)))
        r_voltage = 0.025
        self.R[0, 0] = r_voltage**2.0
        self.R[1, 1] = r_voltage**2.0

        self.KalmanGain, self.Q_steady = controls.kalman(
            A=self.A, B=self.B, C=self.C, Q=self.Q, R=self.R)
        self.L = self.A * self.KalmanGain

        self.U_max = numpy.matrix([[12.0], [12.0]])
        self.U_min = numpy.matrix([[-12.0], [-12.0]])

        self.InitializeState()

class IntegralArm(Arm):

    def __init__(self, name="IntegralArm", J=None):
        super(IntegralArm, self).__init__(name=name, J=J)

        self.A_continuous_unaugmented = self.A_continuous
        self.B_continuous_unaugmented = self.B_continuous

        self.A_continuous = numpy.matrix(numpy.zeros((6, 6)))
        self.A_continuous[0:4, 0:4] = self.A_continuous_unaugmented
        self.A_continuous[0:4, 4:6] = self.B_continuous_unaugmented

        self.B_continuous = numpy.matrix(numpy.zeros((6, 2)))
        self.B_continuous[0:4, 0:2] = self.B_continuous_unaugmented

        self.C_unaugmented = self.C
        self.C = numpy.matrix(numpy.zeros((2, 6)))
        self.C[0:2, 0:4] = self.C_unaugmented

        self.A, self.B = self.ContinuousToDiscrete(self.A_continuous,
                                                   self.B_continuous, self.dt)

        q_pos_shoulder = 0.10
        q_vel_shoulder = 0.005
        q_voltage_shoulder = 3.5
        q_pos_shooter = 0.08
        q_vel_shooter = 2.00
        q_voltage_shooter = 1.0
        self.Q = numpy.matrix(numpy.zeros((6, 6)))
        self.Q[0, 0] = q_pos_shoulder**2.0
        self.Q[1, 1] = q_vel_shoulder**2.0
        self.Q[2, 2] = q_pos_shooter**2.0
        self.Q[3, 3] = q_vel_shooter**2.0
        self.Q[4, 4] = q_voltage_shoulder**2.0
        self.Q[5, 5] = q_voltage_shooter**2.0

        self.R = numpy.matrix(numpy.zeros((2, 2)))
        r_pos = 0.05
        self.R[0, 0] = r_pos**2.0
        self.R[1, 1] = r_pos**2.0

        self.KalmanGain, self.Q_steady = controls.kalman(
            A=self.A, B=self.B, C=self.C, Q=self.Q, R=self.R)
        self.L = self.A * self.KalmanGain

        self.K_unaugmented = self.K
        self.K = numpy.matrix(numpy.zeros((2, 6)))
        self.K[0:2, 0:4] = self.K_unaugmented
        self.K[0, 4] = 1
        self.K[1, 5] = 1

        self.Kff = numpy.concatenate(
            (self.Kff, numpy.matrix(numpy.zeros((2, 2)))), axis=1)

        self.InitializeState()

class ScenarioPlotter(object):

    def __init__(self):
        # Various lists for graphing things.
        self.t = []
        self.x_shoulder = []
        self.v_shoulder = []
        self.a_shoulder = []
        self.x_hat_shoulder = []
        self.u_shoulder = []
        self.offset_shoulder = []
        self.x_shooter = []
        self.v_shooter = []
        self.a_shooter = []
        self.x_hat_shooter = []
        self.u_shooter = []
        self.offset_shooter = []
        self.goal_x_shoulder = []
        self.goal_v_shoulder = []
        self.goal_x_shooter = []
        self.goal_v_shooter = []

    def run_test(self,
                 arm,
                 end_goal,
                 iterations=200,
                 controller=None,
                 observer=None):
        """Runs the plant with an initial condition and goal.

        Args:
            arm: Arm object to use.
            end_goal: numpy.Matrix[6, 1], end goal state.
            iterations: Number of timesteps to run the model for.
            controller: Arm object to get K from, or None if we should
                use arm.
            observer: Arm object to use for the observer, or None if we should
                use the actual state.
        """

        if controller is None:
            controller = arm

        vbat = 12.0

        if self.t:
            initial_t = self.t[-1] + arm.dt
        else:
            initial_t = 0

        goal = numpy.concatenate((arm.X, numpy.matrix(numpy.zeros((2, 1)))),
                                 axis=0)

        shoulder_profile = TrapezoidProfile(arm.dt)
        shoulder_profile.set_maximum_acceleration(12.0)
        shoulder_profile.set_maximum_velocity(10.0)
        shoulder_profile.SetGoal(goal[0, 0])
        shooter_profile = TrapezoidProfile(arm.dt)
        shooter_profile.set_maximum_acceleration(50.0)
        shooter_profile.set_maximum_velocity(10.0)
        shooter_profile.SetGoal(goal[2, 0])

        U_last = numpy.matrix(numpy.zeros((2, 1)))
        for i in xrange(iterations):
            X_hat = arm.X

            if observer is not None:
                observer.Y = arm.Y
                observer.CorrectObserver(U_last)
                self.offset_shoulder.append(observer.X_hat[4, 0])
                self.offset_shooter.append(observer.X_hat[5, 0])

                X_hat = observer.X_hat
                self.x_hat_shoulder.append(observer.X_hat[0, 0])
                self.x_hat_shooter.append(observer.X_hat[2, 0])

            next_shoulder_goal = shoulder_profile.Update(
                end_goal[0, 0], end_goal[1, 0])
            next_shooter_goal = shooter_profile.Update(end_goal[2, 0],
                                                       end_goal[3, 0])

            next_goal = numpy.concatenate(
                (next_shoulder_goal, next_shooter_goal,
                 numpy.matrix(numpy.zeros((2, 1)))),
                axis=0)
            self.goal_x_shoulder.append(goal[0, 0])
            self.goal_v_shoulder.append(goal[1, 0])
            self.goal_x_shooter.append(goal[2, 0])
            self.goal_v_shooter.append(goal[3, 0])

            ff_U = controller.Kff * (next_goal - observer.A * goal)

            U_uncapped = controller.K * (goal - X_hat) + ff_U
            U = U_uncapped.copy()

            U[0, 0] = numpy.clip(U[0, 0], -vbat, vbat)
            U[1, 0] = numpy.clip(U[1, 0], -vbat, vbat)
            self.x_shoulder.append(arm.X[0, 0])
            self.x_shooter.append(arm.X[2, 0])

            if self.v_shoulder:
                last_v_shoulder = self.v_shoulder[-1]
            else:
                last_v_shoulder = 0
            self.v_shoulder.append(arm.X[1, 0])
            self.a_shoulder.append(
                (self.v_shoulder[-1] - last_v_shoulder) / arm.dt)

            if self.v_shooter:
                last_v_shooter = self.v_shooter[-1]
            else:
                last_v_shooter = 0
            self.v_shooter.append(arm.X[3, 0])
            self.a_shooter.append(
                (self.v_shooter[-1] - last_v_shooter) / arm.dt)

            if i % 40 == 0:
                # Test that if we move the shoulder, the shooter stays perfect.
                #observer.X_hat[0, 0] += 0.20
                #arm.X[0, 0] += 0.20
                pass
            U_error = numpy.matrix([[2.0], [2.0]])
            # Kick it and see what happens.
            #if (initial_t + i * arm.dt) % 0.4 > 0.2:
            #U_error = numpy.matrix([[4.0], [0.0]])
            #else:
            #U_error = numpy.matrix([[-4.0], [0.0]])

            arm.Update(U + U_error)

            if observer is not None:
                observer.PredictObserver(U)

            self.t.append(initial_t + i * arm.dt)
            self.u_shoulder.append(U[0, 0])
            self.u_shooter.append(U[1, 0])

            ff_U -= U_uncapped - U
            goal = controller.A * goal + controller.B * ff_U

            if U[0, 0] != U_uncapped[0, 0]:
                glog.debug('Moving shoulder %s', repr(initial_t + i * arm.dt))
                glog.debug('U error %s', repr(U_uncapped - U))
                glog.debug(
                    'goal change is %s',
                    repr(next_shoulder_goal -
                         numpy.matrix([[goal[0, 0]], [goal[1, 0]]])))
                shoulder_profile.MoveCurrentState(
                    numpy.matrix([[goal[0, 0]], [goal[1, 0]]]))
            if U[1, 0] != U_uncapped[1, 0]:
                glog.debug('Moving shooter %s', repr(initial_t + i * arm.dt))
                glog.debug('U error %s', repr(U_uncapped - U))
                shooter_profile.MoveCurrentState(
                    numpy.matrix([[goal[2, 0]], [goal[3, 0]]]))
            U_last = U
        glog.debug('goal_error %s', repr(end_goal - goal))
        glog.debug('error %s', repr(observer.X_hat - end_goal))

    def Plot(self):
        pylab.subplot(3, 1, 1)
        pylab.plot(self.t, self.x_shoulder, label='x shoulder')
        pylab.plot(self.t, self.goal_x_shoulder, label='goal x shoulder')
        pylab.plot(self.t, self.x_hat_shoulder, label='x_hat shoulder')

        pylab.plot(self.t, self.x_shooter, label='x shooter')
        pylab.plot(self.t, self.x_hat_shooter, label='x_hat shooter')
        pylab.plot(self.t, self.goal_x_shooter, label='goal x shooter')
        pylab.plot(
            self.t,
            map(operator.add, self.x_shooter, self.x_shoulder),
            label='x shooter ground')
        pylab.plot(
            self.t,
            map(operator.add, self.x_hat_shooter, self.x_hat_shoulder),
            label='x_hat shooter ground')
        pylab.legend()

        pylab.subplot(3, 1, 2)
        pylab.plot(self.t, self.u_shoulder, label='u shoulder')
        pylab.plot(
            self.t, self.offset_shoulder, label='voltage_offset shoulder')
        pylab.plot(self.t, self.u_shooter, label='u shooter')
        pylab.plot(self.t, self.offset_shooter, label='voltage_offset shooter')
        pylab.legend()

        pylab.subplot(3, 1, 3)
        pylab.plot(self.t, self.a_shoulder, label='a_shoulder')
        pylab.plot(self.t, self.a_shooter, label='a_shooter')
        pylab.legend()

        pylab.show()

def main(argv):
    argv = FLAGS(argv)
    glog.init()

    scenario_plotter = ScenarioPlotter()

    J_accelerating = 18
    J_decelerating = 7

    arm = Arm(name='AcceleratingArm', J=J_accelerating)
    arm_integral_controller = IntegralArm(
        name='AcceleratingIntegralArm', J=J_accelerating)
    arm_observer = IntegralArm()

    # Test moving the shoulder with constant separation.
    initial_X = numpy.matrix([[0.0], [0.0], [0.0], [0.0], [0.0], [0.0]])
    R = numpy.matrix([
        [numpy.pi / 2.0],
        [0.0],
        [0.0],  #[numpy.pi / 2.0],
        [0.0],
        [0.0],
        [0.0]
    ])
    arm.X = initial_X[0:4, 0]
    arm_observer.X = initial_X

    scenario_plotter.run_test(
        arm=arm,
        end_goal=R,
        iterations=300,
        controller=arm_integral_controller,
        observer=arm_observer)

    if len(argv) != 5:
        glog.fatal(
            'Expected .h file name and .cc file name for the wrist and integral wrist.'
        )
    else:
        namespaces = ['y2016', 'control_loops', 'superstructure']
        decelerating_arm = Arm(name='DeceleratingArm', J=J_decelerating)
        loop_writer = control_loop.ControlLoopWriter(
            'Arm', [arm, decelerating_arm], namespaces=namespaces)
        loop_writer.Write(argv[1], argv[2])

        decelerating_integral_arm_controller = IntegralArm(
            name='DeceleratingIntegralArm', J=J_decelerating)

        integral_loop_writer = control_loop.ControlLoopWriter(
            'IntegralArm',
            [arm_integral_controller, decelerating_integral_arm_controller],
            namespaces=namespaces)
        integral_loop_writer.AddConstant(
            control_loop.Constant("kV_shoulder", "%f",
                                  arm_integral_controller.shoulder_Kv))
        integral_loop_writer.Write(argv[3], argv[4])

    if FLAGS.plot:
        scenario_plotter.Plot()

if __name__ == '__main__':
    sys.exit(main(sys.argv))

ext: py

reason: import numpy


text:
from song_data import songs
import numpy as np

# Define percentile and answer here:
percentile = np.quantile(songs, 0.32)
answer = "below"

#Ignore the code below here:
try:
  print("Your percentile is " + str(percentile) + "\n")
except NameError:
  print("You haven't defined percentile")

max_stars_repo_path: Data Scientist Career Path/7. Summary Statistics/5. Quartiles, Quantiles, Interquartile Range/2. Quantiles/4. common.py

max_stars_repo_name: myarist/Codecademy


text:
import numpy as np
from sklearn import mixture
from sklearn.externals import joblib
import matplotlib.pyplot as plt

#ファイル名
csvName_hi = "../result_hi.csv"
csvName_cd = "../result_cd.csv"

#取得する振幅値の数(次元数)
N = 20

#CSVファイルから振幅値の2次元配列取得
def getAmpArray(fileName):
    #CSVファイルをfloatとして読み込み
    data = np.recfromcsv(fileName)

    #numpyのinsert用ダミーデータ
    dummy_data = np.zeros(N)

    array = np.array([dummy_data],dtype=float)
    tmp_array = np.array([],dtype=float)
    n = 1
    m = 0
    #学習用の配列を作成する
    for i in data :
        for j in i:
            tmp_array = np.insert(tmp_array, m, j)
            m += 1

        array = np.insert(array,n,tmp_array,axis=0)
        tmp_array = np.array([],dtype=float)    #tmp_arrayを初期化
        #print(array[n])    #デバッグ
        n+=1
        m=0

    return array

def makeGMM(componets , samples, gmmFileName):
    # GMMを学習
    gmm = mixture.GMM(componets, covariance_type='full')
    gmm.fit(samples)

    #GMMを保存
    joblib.dump(gmm, gmmFileName)

    return gmm

def debug_gmm(gmm) :
    # 結果を表示
    print ("*** weights")   #混合係数
    print (gmm.weights_)

    #print ("*** means")     #平均ベクトル
    #print (gmm.means_)

    #print ("*** covars")    #共分散行列
    #print (gmm.covars_)

    # 0番目のコンポーネントの共分散行列を描画
    plt.imshow(gmm.covars_[0])
    plt.show()

if __name__ == '__main__':
    #CSVファイルから振幅値を取得
    print("振幅値を読み込んでいます。")
    array = getAmpArray(csvName_cd) #変換元
    array2= getAmpArray(csvName_hi) #変換先
    print("振幅値を読み込みました。")

    print("ベクトルを結合しています。\n")
    Z = np.hstack((array,array2))

    gmm = makeGMM(30, Z, "gmm")

    debug_gmm(gmm)


text:
# -*- coding: utf-8 -*-
## @package ivf.core.sfs.colormap_estimation
#
#  ivf.core.sfs.colormap_estimation utility package.
#  @author      tody
#  @date        2016/02/09

import numpy as np
import cv2
import matplotlib.pyplot as plt

from scipy.interpolate.interpolate import interp1d
from scipy.interpolate.fitpack2 import UnivariateSpline

from ivf.cv.image import luminance
from ivf.np.norm import normVectors
from scipy.interpolate.rbf import Rbf

class ColorMapEstimation:
    def __init__(self, Cs, Is=None, num_samples=2000):
        self._Cs = Cs
        self._I0s = np.average(Cs, axis=1)
        if Is is None:
            Is = np.array(self._I0s)
        self._Is = Is
        self._map = None
        self._map_size = 512
        self._num_samples = num_samples

        self._compute()

    def setBrightness(self, Is, update=True):
        self._Is = Is

        if update:
            self._compute()

    def setColor(self, Cs, update=True):
        self._Cs = Cs

        if update:
            self._compute()

    def shading(self, Is):
        I_ids = self._I_ids(Is, self._Iminmax)
        Cs = self._map[I_ids, :]
        return Cs

    def illumination(self, Cs):
        return self._illuminationFromColorDifference(Cs)
        # return self._illuminationFromIrange(Cs)

    def _illuminationFromColorDifference(self, Cs):
        I0s = np.average(Cs, axis=1)

        I0_ids = self._I_ids(I0s)

        C_map = np.zeros((self._map_size, Cs.shape[1]))
        hist = np.zeros((self._map_size))
        C_map[I0_ids, :] += Cs[:, :]
        hist[I0_ids] += 1.0

        hist_positive = hist > 0

        for ci in xrange(3):
            C_map[hist_positive, ci] *= 1.0 / hist[hist_positive]

        I_map = np.zeros((self._map_size))

        I_min, I_max = self._Iminmax

        Ims = np.linspace(I_min, I_max, num=self._map_size)

        for ci in xrange(3):
            C_map[:, ci] = Rbf(Ims[hist_positive], C_map[hist_positive, ci], smooth=0.0005)(Ims)

        sigma = 0.02
        for mi in xrange(self._map_size):
            c = C_map[mi]
            dc = normVectors(self._map - c)

            dc_i = np.argmin(dc)
            wc = np.exp(- (dc ** 2) / (sigma ** 2))

            I = np.dot(wc, Ims) / np.sum(wc)

            #I_map[mi] = I_min + (I_max - I_min) * dc_i / float(self._map_size - 1)
            I_map[mi] = I

        Im_max = 0.0
        for mi in xrange(self._map_size):
            Im_max = max(I_map[mi], Im_max)
            I_map[mi] = Im_max

        I_map = Rbf(Ims, I_map, smooth=0.0005)(Ims)

#         I_map[np.max(I0_ids):] = I_max

        return I_map[I0_ids]

    def _illuminationFromIrange(self, Cs):
        I0s = np.average(Cs, axis=1)
        print "I0 min max", np.min(I0s), np.max(I0s)

        I0_ids = self._I_ids(I0s, self._Im_minmax)

        hist = np.zeros((self._map_size))
        hist[I0_ids] += 1.0

        hist /= np.max(hist)

#         plt.plot(np.arange(self._map_size), hist)
#         plt.show()

        I_map = 0.5 * (self._I_map_min + self._I_map_max)
        #I_map = self._I_map_max
        I_min, I_max = self._Iminmax

        I_map[np.min(I0_ids)] = I_min
        I_map[np.max(I0_ids)] = I_max

        return I_map[I0_ids]

    def mapImage(self, image_size=(256, 256)):
        return cv2.resize(self._map.reshape(1, self._map_size, -1), image_size)

    def _I_ids(self, Is, I_minmax=None):
        if I_minmax is None:
            I_minmax = np.min(Is), np.max(Is)
        I_min, I_max = I_minmax
        I_ids = np.int32((self._map_size - 1) * (Is - I_min) / (I_max - I_min))
        I_ids = np.clip(I_ids, 0, self._map_size - 1)
        return I_ids

    def _computeIrange(self):
        self._I_map_min = np.zeros((self._map_size))
        self._I_map_max = np.zeros((self._map_size))

        hist = np.zeros((self._map_size))

        Is = np.average(self._map, axis=1)

        print "Im min max", np.min(Is), np.max(Is)

        self._Im_minmax = np.min(Is), np.max(Is)

        I_ids = self._I_ids(Is)

        hist[I_ids] += 1.0

        hist_all = np.sum(hist)

        hist_sum = 0.0

        I_min, I_max = self._Iminmax

        for mi in xrange(self._map_size):
            t = hist_sum / hist_all
            self._I_map_min[mi] = I_min + t * (I_max - I_min)
            hist_sum += hist[mi]

            t = (hist_sum - 1.0) / hist_all
            self._I_map_max[mi] = I_min + t * (I_max - I_min)

#         Ims = np.linspace(I_min, I_max, num=self._map_size)
#         self._I_map_max = Rbf(Ims, self._I_map_max, smooth=5)(Ims)
#         self._I_map_min = Rbf(Ims, self._I_map_min, smooth=5)(Ims)

    def _compute(self):
        # self.__computeByPixelList()
        self.__computeByHistogram()
        self._computeIrange()

    def _rbf(self, Is, Cs, smooth=0.00005):
        rbf_list = []
        for ci in xrange(3):
            rbf_list.append(Rbf(Is, Cs[:, ci], smooth=smooth))

        def f(Is_new):
            Cs_new = np.zeros((len(Is_new), Cs.shape[1]))
            for ci in xrange(3):
                Cs_new[:, ci] = rbf_list[ci](Is_new)
            return Cs_new
        return f

    def __computeByHistogram(self):
        sample_ids = np.random.randint(len(self._Is) - 1, size=self._num_samples)
        Is = self._Is[sample_ids]
        I0s = self._I0s[sample_ids]
        Cs = self._Cs[sample_ids]

        I_min, I_max = np.min(Is), np.max(Is)
        self._Iminmax = I_min, I_max

        C_order = np.argsort(I0s)
        Cs_sort = Cs[C_order]
        Is_sort = np.sort(Is)

        hist = np.zeros((self._map_size))
        Cs_avg = np.zeros((self._map_size, Cs.shape[1]))

        I_ids = self._I_ids(Is_sort)
        Cs_avg[I_ids, :] += Cs_sort[:, :]
        hist[I_ids] += 1.0
        hist_positive = hist > 0

        for ci in xrange(3):
            Cs_avg[hist_positive, ci] *= 1.0 / hist[hist_positive]

        self._map = np.zeros((self._map_size, Cs.shape[1]))

        hist_positive = np.where(hist > 0)[0]

        mi = 0
        for hi in hist_positive:

            while mi < hi:
                self._map[mi, :] = Cs_avg[hi, :]
                mi += 1

        while mi < self._map_size:
            self._map[mi, :] = Cs_avg[hist_positive[-1], :]
            mi += 1

        Is_new = np.arange(self._map_size)
        M = self._rbf(Is_new, self._map, smooth=0.00005)
        #self._map = M(Is_new)
        self._map = np.clip(self._map, 0.0, 1.0)

#         Is_avg = np.where(hist > 0)[0]
#         M = []
#
#         for ci in xrange(3):
#             M.append(Rbf(Is_avg, Cs_avg[hist_positive, ci], smooth=0.005))
#
#         for ci in xrange(3):
#             self._map[:, ci] = M[ci](np.arange(self._map_size))

    def __computeByPixelList(self):
        sample_ids = np.random.randint(len(self._Is) - 1, size=self._num_samples)
        Is = self._Is[sample_ids]
        I0s = self._I0s[sample_ids]
        Cs = self._Cs[sample_ids]

        I_min, I_max = np.min(Is), np.max(Is)
        self._Iminmax = I_min, I_max

        C_order = np.argsort(I0s)
        Cs_sort = Cs[C_order]
        Is_sort = np.sort(Is)

        self._map = np.zeros((self._map_size, Cs.shape[1]))
        I_i = 0
        for mi in xrange(self._map_size):
            Im = I_min + (I_max - I_min) * mi / (self._map_size - 1)

            while I_i < len(Is_sort) - 1 and Is_sort[I_i] < Im:
                    I_i += 1

            self._map[mi, :] = Cs_sort[I_i, :]

        self._map = cv2.bilateralFilter(
                                        np.float32(self._map.reshape(1, self._map_size, Cs.shape[1])), 0, 0.1, 2).reshape(self._map_size, Cs.shape[1])


text:
import cv2
import numpy as np

# imporing the two algorithms, SVM and KNN
from sklearn.svm import SVC
from sklearn.neighbors import KNeighborsClassifier as KNN

# some helper functions
from sklearn.model_selection import train_test_split as tts

# for evaluating our trained models
from sklearn.metrics import accuracy_score

# set some constants
N_SAMPLES = 1000
TEST_SIZE = 0.2
APPLE = 0
BANANA = 1

def normalize(data):
    "Takes a list or a list of lists and returns its normalized form"

    return np.interp(data, [0, 255], [-1, 1])

# load the individual datasets
apples_full = np.load('dataset/apple.npy')
bananas_full = np.load('dataset/banana.npy')

# take only N_SAMPLES number of samples from each dataset
apples = apples_full[:N_SAMPLES]
bananas = bananas_full[:N_SAMPLES]

# put them together to build a collective dataset
dataset = np.concatenate((apples, bananas))

# normalize the values
dataset = normalize(dataset)

# make the labels corresponding to the categories
labels = [APPLE] * N_SAMPLES + [BANANA] * N_SAMPLES

# prepare the data, split into training and testing
x_train, x_test, y_train, y_test = tts(dataset,
                                       labels,
                                       test_size=TEST_SIZE)

# choose and set the classifier algorithm
clf = KNN()

# train it
clf.fit(x_train, y_train)

# start predicting!
preds = clf.predict(x_test)

# evauluate the model
accuracy = accuracy_score(y_test, preds)
print("Accuracy:", accuracy)


text:
import numpy as np
from .distribution import Distribution
from scipy.stats import multivariate_normal

class GaussianDistribution(Distribution):
    """
    Gaussian distribution with fixed covariance matrix. The parameters
    vector represents only the mean.

    """
    def __init__(self, mu, sigma):
        """
        Constructor.

        Args:
            mu (np.ndarray): initial mean of the distribution;
            sigma (np.ndarray): covariance matrix of the distribution.

        """
        self._mu = mu
        self._sigma = sigma
        self._inv_sigma = np.linalg.inv(sigma)

        self._add_save_attr(
            _mu='numpy',
            _sigma='numpy',
            _inv_sigma='numpy'
        )

    def sample(self):
        return np.random.multivariate_normal(self._mu, self._sigma)

    def log_pdf(self, theta):
        return multivariate_normal.logpdf(theta, self._mu, self._sigma)

    def __call__(self, theta):
        return multivariate_normal.pdf(theta, self._mu, self._sigma)

    def entropy(self):
        return 0.5 * np.log(np.linalg.det(2*np.pi*np.e*self._sigma))

    def mle(self, theta, weights=None):
        if weights is None:
            self._mu = np.mean(theta, axis=0)
        else:
            self._mu = weights.dot(theta) / np.sum(weights)

    def diff_log(self, theta):
        delta = theta - self._mu
        g = self._inv_sigma.dot(delta)

        return g

    def get_parameters(self):
        return self._mu

    def set_parameters(self, rho):
        self._mu = rho

    @property
    def parameters_size(self):
        return len(self._mu)

class GaussianDiagonalDistribution(Distribution):
    """
    Gaussian distribution with diagonal covariance matrix. The parameters
    vector represents the mean and the standard deviation for each dimension.

    """
    def __init__(self, mu, std):
        """
        Constructor.

        Args:
            mu (np.ndarray): initial mean of the distribution;
            std (np.ndarray): initial vector of standard deviations for each
                variable of the distribution.

        """
        assert(len(std.shape) == 1)
        self._mu = mu
        self._std = std

        self._add_save_attr(
            _mu='numpy',
            _std='numpy'
        )

    def sample(self):
        sigma = np.diag(self._std**2)
        return np.random.multivariate_normal(self._mu, sigma)

    def log_pdf(self, theta):
        sigma = np.diag(self._std ** 2)
        return multivariate_normal.logpdf(theta, self._mu, sigma)

    def __call__(self, theta):
        sigma = np.diag(self._std ** 2)
        return multivariate_normal.pdf(theta, self._mu, sigma)

    def entropy(self):
        return 0.5 * np.log(np.product(2*np.pi*np.e*self._std**2))

    def mle(self, theta, weights=None):
        if weights is None:
            self._mu = np.mean(theta, axis=0)
            self._std = np.std(theta, axis=0)
        else:
            sumD = np.sum(weights)
            sumD2 = np.sum(weights**2)
            Z = sumD - sumD2 / sumD

            self._mu = weights.dot(theta) / sumD

            delta2 = (theta - self._mu)**2
            self._std = np.sqrt(weights.dot(delta2) / Z)

    def diff_log(self, theta):
        n_dims = len(self._mu)

        sigma = self._std**2

        g = np.empty(self.parameters_size)

        delta = theta - self._mu

        g_mean = delta / sigma
        g_cov = delta**2 / (self._std**3) - 1 / self._std

        g[:n_dims] = g_mean
        g[n_dims:] = g_cov
        return g

    def get_parameters(self):
        rho = np.empty(self.parameters_size)
        n_dims = len(self._mu)

        rho[:n_dims] = self._mu
        rho[n_dims:] = self._std

        return rho

    def set_parameters(self, rho):
        n_dims = len(self._mu)
        self._mu = rho[:n_dims]
        self._std = rho[n_dims:]

    @property
    def parameters_size(self):
        return 2 * len(self._mu)

class GaussianCholeskyDistribution(Distribution):
    """
    Gaussian distribution with full covariance matrix. The parameters
    vector represents the mean and the Cholesky decomposition of the
    covariance matrix. This parametrization enforce the covariance matrix to be
    positive definite.

    """
    def __init__(self, mu, sigma):
        """
        Constructor.

        Args:
            mu (np.ndarray): initial mean of the distribution;
            sigma (np.ndarray): initial covariance matrix of the distribution.

        """
        self._mu = mu
        self._chol_sigma = np.linalg.cholesky(sigma)

        self._add_save_attr(
            _mu='numpy',
            _chol_sigma='numpy'
        )

    def sample(self):
        sigma = self._chol_sigma.dot(self._chol_sigma.T)
        return np.random.multivariate_normal(self._mu, sigma)

    def log_pdf(self, theta):
        sigma = self._chol_sigma.dot(self._chol_sigma.T)
        return multivariate_normal.logpdf(theta, self._mu, sigma)

    def __call__(self, theta):
        sigma = self._chol_sigma.dot(self._chol_sigma.T)
        return multivariate_normal.pdf(theta, self._mu, sigma)

    def entropy(self):
        std = np.diag(self._chol_sigma)
        return 0.5 * np.log(np.product(2*np.pi*np.e*std**2))

    def mle(self, theta, weights=None):
        if weights is None:
            self._mu = np.mean(theta, axis=0)
            sigma = np.cov(theta, rowvar=False)
        else:
            sumD = np.sum(weights)
            sumD2 = np.sum(weights**2)
            Z = sumD - sumD2 / sumD

            self._mu = weights.dot(theta) / sumD

            delta = theta - self._mu

            sigma = delta.T.dot(np.diag(weights)).dot(delta) / Z

        self._chol_sigma = np.linalg.cholesky(sigma)

    def diff_log(self, theta):
        n_dims = len(self._mu)
        inv_chol = np.linalg.inv(self._chol_sigma)
        inv_sigma = inv_chol.T.dot(inv_chol)

        g = np.empty(self.parameters_size)

        delta = theta - self._mu
        g_mean = inv_sigma.dot(delta)

        delta_a = np.reshape(delta, (-1, 1))
        delta_b = np.reshape(delta, (1, -1))

        S = inv_chol.dot(delta_a).dot(delta_b).dot(inv_sigma)

        g_cov = S - np.diag(np.diag(inv_chol))

        g[:n_dims] = g_mean
        g[n_dims:] = g_cov.T[np.tril_indices(n_dims)]

        return g

    def get_parameters(self):
        rho = np.empty(self.parameters_size)
        n_dims = len(self._mu)

        rho[:n_dims] = self._mu
        rho[n_dims:] = self._chol_sigma[np.tril_indices(n_dims)]

        return rho

    def set_parameters(self, rho):
        n_dims = len(self._mu)
        self._mu = rho[:n_dims]
        self._chol_sigma[np.tril_indices(n_dims)] = rho[n_dims:]

    @property
    def parameters_size(self):
        n_dims = len(self._mu)

        return 2 * n_dims + (n_dims * n_dims - n_dims) // 2

max_stars_repo_name: AmmarFahmy/mushroom-rl

max_issues_repo_name: AmmarFahmy/mushroom-rl

max_forks_repo_name: AmmarFahmy/mushroom-rl


text:
from PIL import Image
import numpy as np
from tensorboardX import SummaryWriter
from tensorboardX import utils

class UnityModel(object):
	def __init__(self):
		self.writer = SummaryWriter()

	def save_image(self, title, nparray_of_pixels, iter):
		#nparray_of_pixels = np.transpose(nparray_of_pixels,(2,0,1))
		#formatted = (nparray_of_pixels * 255 / np.max(nparray_of_pixels)).astype('uint8')
		self.writer.add_image(title, utils.make_grid(nparray_of_pixels, ncols=4), iter)

max_stars_repo_name: Unity-Technologies/Keras-GAN

max_issues_repo_name: Unity-Technologies/Keras-GAN

max_forks_repo_name: Unity-Technologies/Keras-GAN


text:
"""
Test cases for the regi0.geographic.local.get_layer_field function.
"""
import geopandas as gpd
import numpy as np
import pandas as pd

from regi0.geographic.local import get_layer_field

def test_geodataframe(records, data_path):
    countries = gpd.read_file(
        data_path.joinpath("gpkg/admin0.gpkg"), layer="admin0_2018"
    )
    result = get_layer_field(records, countries, field="ISO_A2")
    expected = pd.Series(
        [
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "VE",
            "CO",
            "VE",
            "VE",
        ]
    )
    pd.testing.assert_series_equal(result, expected, check_names=False)

def test_geojson(records, data_path):
    result = get_layer_field(
        records,
        data_path.joinpath("geojson/urban.geojson"),
        field="nom_cpob",
    )
    expected = pd.Series(
        [
            np.nan,
            np.nan,
            np.nan,
            np.nan,
            np.nan,
            np.nan,
            np.nan,
            "SAN GIL",
            np.nan,
            np.nan,
            np.nan,
            "SOATA",
            np.nan,
            np.nan,
            np.nan,
            np.nan,
            np.nan,
            np.nan,
            np.nan,
            np.nan,
            np.nan,
            np.nan,
        ]
    )
    pd.testing.assert_series_equal(result, expected, check_names=False)

def test_geopackage(records, data_path):
    result = get_layer_field(
        records,
        data_path.joinpath("gpkg/admin0.gpkg"),
        layer="admin0_2018",
        field="ISO_A2",
    )
    expected = pd.Series(
        [
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "CO",
            "VE",
            "CO",
            "VE",
            "VE",
        ]
    )
    pd.testing.assert_series_equal(result, expected, check_names=False)

def test_shapefile(records, data_path):
    result = get_layer_field(
        records,
        data_path.joinpath("shp/admin1_2003.shp"),
        field="dptos",
    )
    expected = pd.Series(
        [
            "BOYACA",
            "BOYACA",
            "BOYACA",
            "BOYACA",
            "SANTANDER",
            "SANTANDER",
            "SANTANDER",
            "BOYACA",
            "SANTANDER",
            "SANTANDER",
            "BOYACA",
            "BOYACA",
            "SANTANDER",
            "SANTANDER",
            "BOYACA",
            "SANTANDER",
            "BOYACA",
            "SANTANDER",
            np.nan,
            "BOYACA",
            np.nan,
            np.nan,
        ]
    )
    pd.testing.assert_series_equal(result, expected, check_names=False)


text:
import time
import matplotlib.pyplot as plt
from sklearn.datasets import load_iris
from yellowbrick.features import ParallelCoordinates
import pandas as pd
import numpy as np

def plot_speedup(trials=5, factors=np.arange(1, 11)):

    def pcoords_time(X, y, fast=True):
        _, ax = plt.subplots()
        oz = ParallelCoordinates(fast=fast, ax=ax)

        start = time.time()
        oz.fit_transform(X, y)
        delta = time.time() - start

        plt.cla()        # clear current axis
        plt.clf()        # clear current figure
        plt.close("all") # close all existing plots

        return delta

    def pcoords_speedup(X, y):
        fast_time = pcoords_time(X, y, fast=True)
        slow_time = pcoords_time(X, y, fast=False)

        return slow_time / fast_time

    data = load_iris()

    speedups = []
    variance = []

    for factor in factors:
        X = np.repeat(data.data, factor, axis=0)
        y = np.repeat(data.target, factor, axis=0)

        local_speedups = []
        for trial in range(trials):
            local_speedups.append(pcoords_speedup(X, y))

        local_speedups = np.array(local_speedups)
        speedups.append(local_speedups.mean())
        variance.append(local_speedups.std())

    speedups = np.array(speedups)
    variance = np.array(variance)

    series = pd.Series(speedups, index=factors)
    _, ax = plt.subplots(figsize=(9,6))
    series.plot(ax=ax, marker='o', label="speedup factor", color='b')

    # Plot one standard deviation above and below the mean
    ax.fill_between(
        factors, speedups - variance, speedups + variance, alpha=0.25,
        color='b',
    )

    ax.set_ylabel("speedup factor")
    ax.set_xlabel("dataset size (number of repeats in Iris dataset)")
    ax.set_title("Speed Improvement of Fast Parallel Coordinates")
    plt.savefig("images/fast_parallel_coordinates_speedup_benchmark.png")

if __name__ == '__main__':
    plot_speedup()

max_stars_repo_name: xingularity/yellowbrick

max_issues_repo_name: xingularity/yellowbrick

max_forks_repo_name: xingularity/yellowbrick


text:
import numpy as np
import sympy
from sympy import Matrix

DOF = 2

def name_ind(i):
    if i >=0 and i < DOF:
        return 'c1'
    elif i >= DOF and i < 2*DOF:
        return 'c2'
    elif i >= 2*DOF and i < 3*DOF:
        return 'c3'
    else:
        raise

sympy.var('A, le, xi')
sympy.var('rho, E, nu, c, s')

N1 = (1-xi)/2
N2 = (1+xi)/2

Nu = Matrix([[N1, 0, N2, 0]])
Nv = Matrix([[0, N1, 0, N2]])

# Constitutive linear stiffness matrix
#c
R = Matrix([[c, s, 0, 0],
            [-s, c, 0, 0],
            [0, 0, c, s],
            [0, 0, -s, c]])

num_nodes = 2

BLu = Matrix([[(2/le)*N1.diff(xi), 0, (2/le)*N2.diff(xi), 0]])

Ke = A*E/le*Matrix([[1, 0, -1, 0],
                    [0, 0, 0, 0],
                    [-1, 0, 1, 0],
                    [0, 0, 0, 0]])

Me = (le/2)*sympy.integrate(rho*A*(Nu.T*Nu + Nv.T*Nv), (xi, -1, +1))
Me_lumped = A*rho*le/2*Matrix(
                  [[1, 0, 0, 0],
                   [0, 1, 0, 0],
                   [0, 0, 1, 0],
                   [0, 0, 0, 1]])

print('printing Me')
for ind, val in np.ndenumerate(Me):
    if val == 0:
        continue
    i, j = ind
    si = name_ind(i)
    sj = name_ind(j)
    print('    Me[%d+%s, %d+%s]' % (i%DOF, si, j%DOF, sj), '+=', Me[ind])
print('printing Me_lumped')
for ind, val in np.ndenumerate(Me_lumped):
    if val == 0:
        continue
    i, j = ind
    si = name_ind(i)
    sj = name_ind(j)
    print('    Me[%d+%s, %d+%s]' % (i%DOF, si, j%DOF, sj), '+=', Me_lumped[ind])

K = sympy.zeros(num_nodes*DOF, num_nodes*DOF)
M = sympy.zeros(num_nodes*DOF, num_nodes*DOF)
M_lumped = sympy.zeros(num_nodes*DOF, num_nodes*DOF)

# global
K[:, :] = R.T*Ke*R
M[:, :] = R.T*Me*R
M_lumped[:, :] = R.T*Me_lumped*R

# K represents the global stiffness matrix
# in case we want to apply coordinate transformations

print('printing K')
for ind, val in np.ndenumerate(K):
    i, j = ind
    si = name_ind(i)
    sj = name_ind(j)
    print('    K[%d+%s, %d+%s]' % (i%DOF, si, j%DOF, sj), '+=', K[ind])
print('printing M')
for ind, val in np.ndenumerate(M):
    if val == 0:
        continue
    i, j = ind
    si = name_ind(i)
    sj = name_ind(j)
    print('    M[%d+%s, %d+%s]' % (i%DOF, si, j%DOF, sj), '+=', M[ind])
print('printing M_lumped')
for ind, val in np.ndenumerate(M_lumped):
    if val == 0:
        continue
    i, j = ind
    si = name_ind(i)
    sj = name_ind(j)
    print('    M[%d+%s, %d+%s]' % (i%DOF, si, j%DOF, sj), '+=', M_lumped[ind])

reason: import numpy,import sympy,from sympy


text:
import nibabel as nib
import numpy as np

def tract2tck(data, filename, affine_to_rasmm=None):
    if not affine_to_rasmm:
        affine_to_rasmm = np.eye(4,4)

    t_gram = nib.streamlines.Tractogram(data, affine_to_rasmm=affine_to_rasmm)
    tck_out = nib.streamlines.TckFile(t_gram)
    tck_out.save(filename)

max_issues_repo_name: BioMedAnalysis/latenttrack

max_forks_repo_name: BioMedAnalysis/latenttrack


text:
# coding: utf-8
import numpy as np
import pandas as pd
import pickle

def excel_to_pkl():

    df = pd.read_excel("connection.xlsx")
    print(df)
    connection_s = df.values
    connection = connection_s[:, 1:]
    connection = connection.astype(np.float32)
    print(connection)
    f = open('adj_mx_subway.pkl', 'wb+')
    sensor_ids, sensor_id_to_ind = 1, 2
    pickle.dump([sensor_ids, sensor_id_to_ind, connection], f)
    f.close()
excel_to_pkl()


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

from helpers import assert_equality

def plot():
    da = np.zeros((3, 3))
    da[:2, :2] = 1.0

    fig = plt.figure()
    ax = plt.gca()

    im = ax.imshow(da, cmap="viridis")
    plt.colorbar(im, aspect=5, shrink=0.5)
    return fig

def test():
    assert_equality(plot, "test_fancy_colorbar_reference.tex")
    return

max_stars_repo_name: eric-wieser/tikzplotlib

max_issues_repo_name: eric-wieser/tikzplotlib

max_forks_repo_name: eric-wieser/tikzplotlib


text:
#IMPORTACIONES
from binance_f import RequestClient
from binance_f.model import *
from binance_f.constant.test import *
from binance_f.base.printobject import *

import pandas as pd
import numpy as np
import plotly.express as px
import time
import datetime as dt
import os

#CONEXIÓN CON LA API DE BINANCE
request_client = RequestClient(api_key=g_api_key, secret_key=g_secret_key)

#LISTA DE LAS MONEDAS DE FUTUROS
symbols = ['1000SHIBUSDT', '1INCHUSDT', 'AAVEUSDT', 'ADAUSDT', 'AKROUSDT', 'ALGOUSDT', 'ALICEUSDT', 'ALPHAUSDT', 'ANKRUSDT', 'ATOMUSDT', 'AVAXUSDT', 'AXSUSDT', 'BAKEUSDT', 'BALUSDT', 'BANDUSDT', 'BATUSDT', 'BCHUSDT', 'BELUSDT', 'BLZUSDT', 'BNBUSDT', 'BTCUSDT', 'BTSUSDT', 'BTTUSDT', 'BZRXUSDT', 'CELRUSDT', 'CHRUSDT', 'CHZUSDT', 'COMPUSDT', 'COTIUSDT', 'CRVUSDT', 'CTKUSDT', 'CVCUSDT', 'DASHUSDT', 'DEFIUSDT', 'DENTUSDT', 'DGBUSDT', 'DODOUSDT', 'DOGEUSDT', 'DOTUSDT', 'EGLDUSDT', 'ENJUSDT', 'EOSUSDT', 'ETCUSDT', 'ETHUSDT', 'FILUSDT', 'FLMUSDT', 'FTMUSDT', 'GRTUSDT', 'GTCUSDT', 'HBARUSDT', 'HNTUSDT', 'HOTUSDT', 'ICPUSDT', 'ICXUSDT', 'IOSTUSDT', 'IOTAUSDT', 'KAVAUSDT', 'KEEPUSDT', 'KNCUSDT', 'KSMUSDT', 'LINAUSDT', 'LINKUSDT', 'LITUSDT', 'LRCUSDT', 'LTCUSDT', 'LUNAUSDT', 'MANAUSDT', 'MATICUSDT', 'MKRUSDT', 'MTLUSDT', 'NEARUSDT', 'NEOUSDT', 'NKNUSDT', 'OCEANUSDT', 'OGNUSDT', 'OMGUSDT', 'ONEUSDT', 'ONTUSDT', 'QTUMUSDT', 'REEFUSDT', 'RENUSDT', 'RLCUSDT', 'RSRUSDT', 'RUNEUSDT', 'RVNUSDT', 'SANDUSDT', 'SFPUSDT', 'SKLUSDT', 'SNXUSDT', 'SOLUSDT', 'SRMUSDT', 'STMXUSDT', 'STORJUSDT', 'SUSHIUSDT', 'SXPUSDT', 'THETAUSDT', 'TOMOUSDT', 'TRBUSDT', 'TRXUSDT', 'UNFIUSDT', 'UNIUSDT', 'VETUSDT', 'WAVESUSDT', 'XEMUSDT', 'XLMUSDT', 'XMRUSDT', 'XRPUSDT', 'XTZUSDT', 'YFIIUSDT', 'YFIUSDT', 'ZECUSDT', 'ZENUSDT', 'ZILUSDT', 'ZRXUSDT']

def getCandles(symbols):

    #DATAFRAME CON LA INFORMACIÓN
	df = pd.DataFrame(columns= ['Symbol', 'Change', 'Signal', 'Volume'])

    #ARRAYS PARA GUARDAR LA DATA
	coin, change, signal, vol = [], [], [], []

    #RECORRER EL ARRAY DE MONEDAS
	for symbol in symbols:

        #OBTENER LA INFORMACIÓN DE LAS VELAS DE LOS ÚLTIMOS 45 MINUTOS
		candles = request_client.get_candlestick_data(symbol=symbol,interval=CandlestickInterval.MIN1, limit=45)

        #NOMBRE DE LA MONEDA
		sym = symbol[:-4]

        #DATOS BASE DE COMPARACIÓN DE PRECIO MÁXIMO Y MÍNIMO
		highPrice = 0
		lowPrice = 99999999999

        #RECORRER LAS VELAS DE LA MONEDA
		for candle in candles:

            #SI EL PRECIO DE LA VELA ES MAYOR QUE EL ULTIMO PRECIO MAYOR, CAMBIARLO
			if (float(candle.high) > highPrice):

				highPrice = float(candle.high)

            #SI EL PRECIO DE LA VELA ES MENOR QUE EL ULTIMO PRECIO MENOR, CAMBIARLO
			if (float(candle.low) < lowPrice):

				lowPrice = float(candle.low)

        #CALCULANDO EL PORCENTAJE DE CAMBIO DEL MAYOR Y EL MENOR CON RESPECTO AL PRECIO ACTUAL
		changeHigh = float(((float(candles[44].close) - float(highPrice)) / float(highPrice)) * 100)
		changeLow = float(((float(candles[44].close) - float(lowPrice)) / float(lowPrice)) * 100)

        #ANALIZANDO EL MAYOR CAMBIO Y EL TIPO DE SEÑAL
		changeAbs = 0
		sign = ''

		if(abs(changeHigh) > abs(changeLow)):

			changeAbs = abs(changeHigh)
			sign = 'LONG'

		else:

			changeAbs = abs(changeLow)
			sign = 'SHORT'

        #OBTENIENDO EL VOLUMEN DE LAS ÚLTIMAS 24 HORAS DE LA MONEDA
		volume = request_client.get_ticker_price_change_statistics(symbol=symbol)[0].quoteVolume

        #AGREGAR LOS DATOS A LOS ARRAYS DE DATOS
		coin.append(sym)
		change.append(changeAbs)
		signal.append(sign)
		vol.append(volume)

    #ASIGNAR LOS DATOS AL DATAFRAME
	df['Symbol'] = coin
	df['Change'] = np.array(change).astype(np.float)
	df['Volume'] = np.array(vol).astype(np.float)
	df['Signal'] = sign

    #RETORNAR EL DATAFRAME
	return df

def get_moving_symbols(candles_df):

	df1 = candles_df[candles_df['Volume'] >= 100000000]
	df1 = df1[df1['Change'] > 5]

	df2 = candles_df[candles_df['Volume'] < 100000000]
	df2 = df2[df2['Volume'] > 50000000]
	df2 = df2[df2['Change'] > 7]

	df3 = candles_df[candles_df['Volume'] < 50000000]
	df3 = df3[df3['Change'] > 10]

	df = [df1, df2, df3]
	df = pd.concat(df)

	return df

def main():

	while(True):

		candles_df = getCandles(symbols)
		candles_df_top = get_moving_symbols(candles_df)

		candles_df = candles_df.sort_values(by=['Change'], ascending = False)
		candles_df = candles_df.head(10)

		candles_df_top = candles_df_top.sort_values(by=['Change'], ascending = False)

		print('\nMonedas más movidas')
		print(candles_df)
		print('\nSeñales')
		print(candles_df_top)
		time.sleep(60)

if __name__ == '__main__':
	main()

max_stars_repo_name: Gisellemfc/scanner-crypto-futures

max_issues_repo_name: Gisellemfc/scanner-crypto-futures

max_forks_repo_name: Gisellemfc/scanner-crypto-futures


text:
# -*- coding:utf-8 -*-
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import argparse
import ast
import copy
import math
import os
import time

from paddle.fluid.core import AnalysisConfig, create_paddle_predictor, PaddleTensor
from paddlehub.common.logger import logger
from paddlehub.module.module import moduleinfo, runnable, serving
from PIL import Image
import cv2
import numpy as np
import paddle.fluid as fluid
import paddlehub as hub

from tools.infer.utility import draw_boxes, base64_to_cv2
from tools.infer.predict_det import TextDetector

@moduleinfo(
    name="ocr_det",
    version="1.0.0",
    summary="ocr detection service",
    author="paddle-dev",
    author_email="paddle-dev@baidu.com",
    type="cv/text_recognition")
class OCRDet(hub.Module):
    def _initialize(self, use_gpu=False):
        """
        initialize with the necessary elements
        """
        from ocr_det.params import read_params
        cfg = read_params()

        cfg.use_gpu = use_gpu
        if use_gpu:
            try:
                _places = os.environ["CUDA_VISIBLE_DEVICES"]
                int(_places[0])
                print("use gpu: ", use_gpu)
                print("CUDA_VISIBLE_DEVICES: ", _places)
                cfg.gpu_mem = 8000
            except:
                raise RuntimeError(
                    "Environment Variable CUDA_VISIBLE_DEVICES is not set correctly. If you wanna use gpu, please set CUDA_VISIBLE_DEVICES via export CUDA_VISIBLE_DEVICES=cuda_device_id."
                )
        cfg.ir_optim = True

        self.text_detector = TextDetector(cfg)

    def read_images(self, paths=[]):
        images = []
        for img_path in paths:
            assert os.path.isfile(
                img_path), "The {} isn't a valid file.".format(img_path)
            img = cv2.imread(img_path)
            if img is None:
                logger.info("error in loading image:{}".format(img_path))
                continue
            images.append(img)
        return images

    def predict(self,
                images=[],
                paths=[],
                draw_img_save='ocr_det_result',
                visualization=False):
        """
        Get the text box in the predicted images.
        Args:
            images (list(numpy.ndarray)): images data, shape of each is [H, W, C]. If images not paths
            paths (list[str]): The paths of images. If paths not images
            draw_img_save (str): The directory to store output images.
            visualization (bool): Whether to save image or not.
        Returns:
            res (list): The result of text detection box and save path of images.
        """

        if images != [] and isinstance(images, list) and paths == []:
            predicted_data = images
        elif images == [] and isinstance(paths, list) and paths != []:
            predicted_data = self.read_images(paths)
        else:
            raise TypeError("The input data is inconsistent with expectations.")

        assert predicted_data != [], "There is not any image to be predicted. Please check the input data."

        all_results = []
        for img in predicted_data:
            result = {'save_path': ''}
            if img is None:
                logger.info("error in loading image")
                result['data'] = []
                all_results.append(result)
                continue
            dt_boxes, elapse = self.text_detector(img)
            print("Predict time : ", elapse)
            result['data'] = dt_boxes.astype(np.int).tolist()

            if visualization:
                image = Image.fromarray(cv2.cvtColor(img, cv2.COLOR_BGR2RGB))
                draw_img = draw_boxes(image, dt_boxes)
                draw_img = np.array(draw_img)
                if not os.path.exists(draw_img_save):
                    os.makedirs(draw_img_save)
                saved_name = 'ndarray_{}.jpg'.format(time.time())
                save_file_path = os.path.join(draw_img_save, saved_name)
                cv2.imwrite(save_file_path, draw_img[:, :, ::-1])
                print("The visualized image saved in {}".format(save_file_path))
                result['save_path'] = save_file_path

            all_results.append(result)
        return all_results

    @serving
    def serving_method(self, images, **kwargs):
        """
        Run as a service.
        """
        images_decode = [base64_to_cv2(image) for image in images]
        results = self.predict(images_decode, **kwargs)
        return results

if __name__ == '__main__':
    ocr = OCRDet()
    image_path = [
        './doc/imgs/11.jpg',
        './doc/imgs/12.jpg',
    ]
    res = ocr.predict(paths=image_path, visualization=True)
    print(res)

max_stars_repo_name: qiyuezhiguang/PaddleOCR

max_issues_repo_name: qiyuezhiguang/PaddleOCR

max_forks_repo_name: qiyuezhiguang/PaddleOCR


text:
"""
This module tests the central islatu.scan module's Scan and Scan2D classes.
"""

import pytest
from pytest_lazyfixture import lazy_fixture as lazy
import numpy as np
from scipy.interpolate import interp1d

from islatu.background import roi_subtraction, fit_gaussian_1d
from islatu.cropping import crop_to_region
from islatu.scan import Scan2D
from islatu.region import Region

def test_subsample_q_01(scan2d_from_nxs_01: Scan2D):
    """
    Make sure subsample_q deletes the appropriate things. Because it just calls
    remove_data_points, which has already been tested extensively in test_data,
    we only need to check a couple of values to make sure the right qs have been
    deleted an we know that all the other attributes will have been handled
    correctly.
    """
    original_len = len(scan2d_from_nxs_01.theta)
    # Defaults shouldn't change anything.
    scan2d_from_nxs_01.subsample_q()
    assert len(scan2d_from_nxs_01.theta) == original_len

def test_subsample_q_02(scan2d_from_nxs_01: Scan2D):
    """
    Make sure that we can set just an upper bound. Note that this dataset goes
    from 0.025Å to 0.06Å
    """
    q_max = 0.04

    assert max(scan2d_from_nxs_01.q_vectors) > q_max
    scan2d_from_nxs_01.subsample_q(q_max=q_max)
    assert max(scan2d_from_nxs_01.q_vectors) <= q_max

def test_subsample_q_03(scan2d_from_nxs_01: Scan2D):
    """
    Make sure that we can set a lower bound. Note that this dataset goes from
    0.025Å to 0.06Å.
    """
    q_min = 0.04

    assert min(scan2d_from_nxs_01.q_vectors) < q_min
    scan2d_from_nxs_01.subsample_q(q_min=q_min)
    assert min(scan2d_from_nxs_01.q_vectors) >= q_min

def test_subsample_q_04(scan2d_from_nxs_01: Scan2D):
    """
    Test that we can set both lower and upper bounds.
    """
    q_min = 0.032
    q_max = 0.051

    scan2d_from_nxs_01.subsample_q(q_min, q_max)

    assert min(scan2d_from_nxs_01.q_vectors) >= q_min
    assert max(scan2d_from_nxs_01.q_vectors) <= q_max

@pytest.mark.parametrize(
    'scan, transmission',
    [
        (lazy('scan2d_from_nxs_01'), 0.000448426658633058)
    ]
)
def test_transmission_normalisation_intensities(scan: Scan2D, transmission):
    """
    Make sure that we can correct for the attenuation of the beam. The
    transmission values have been manually read from the .nxs file using a GUI.
    """
    intensity_0 = np.copy(scan.intensity)
    scan.transmission_normalisation()

    for i, intensity in enumerate(scan.intensity):
        assert intensity == intensity_0[i]/transmission

@pytest.mark.parametrize(
    'scan, transmission',
    [
        (lazy('scan2d_from_nxs_01'), 0.000448426658633058)
    ]
)
def test_transmission_normalisation_errors(scan: Scan2D, transmission):
    """
    Make sure that we can correct for the attenuation of the beam. The
    transmission values have been manually read from the .nxs file using a GUI.
    This function checks the intensity_e values have been dealt with properly.
    """
    intensity_e_0 = np.copy(scan.intensity_e)
    scan.transmission_normalisation()

    for i, intensity_e in enumerate(scan.intensity_e):
        assert intensity_e == intensity_e_0[i]/transmission

def test_qdcd_name_assumes(parsed_dcd_normalisation_01):
    """
    Takes a parsed DCD normalisation pandas dataframe and makes sure that
    we can find the qdcd data, which is [in]conveniently called qdcd_.
    """
    _, dataframe = parsed_dcd_normalisation_01
    assert "qdcd_" in dataframe
    assert "adc2" in dataframe

def test_qdcd_normalisation_01(scan2d_from_nxs_01: Scan2D, dcd_norm_01_splev):
    """
    Make sure that our qdcd normalisation is doing something, and isn't failing
    silently. (This is a dumb test, but it's really quite hard to test that
    this is working without just rewriting a division by splev).
    """
    intensities_0 = np.copy(scan2d_from_nxs_01.intensity)
    intensities_e_0 = np.copy(scan2d_from_nxs_01.intensity_e)

    scan2d_from_nxs_01.qdcd_normalisation(dcd_norm_01_splev)

    assert (intensities_0 != scan2d_from_nxs_01.intensity).all()
    assert (intensities_e_0 != scan2d_from_nxs_01.intensity_e).all()

def test_qdcd_normalisation_02(scan2d_from_nxs_01: Scan2D, dcd_norm_01_splev,
                               parsed_dcd_normalisation_01):
    """
    Make sure that our nice splev normalisation does something similar to what
    would be achieved using a simple cubic scipy.interpolate.interp1D.
    """

    # First, generate some test intensities by dividing by an interp1D function.
    intensities_0 = np.copy(scan2d_from_nxs_01.intensity)
    intensities_e_0 = np.copy(scan2d_from_nxs_01.intensity_e)

    _, dataframe = parsed_dcd_normalisation_01

    interp = interp1d(dataframe["qdcd_"], dataframe['adc2'], kind='cubic')

    test_intensities = intensities_0 / interp(scan2d_from_nxs_01.q_vectors)
    test_intensities_e = intensities_e_0 / interp(scan2d_from_nxs_01.q_vectors)

    # Now, carry out the qdcd normalisation as normal.
    scan2d_from_nxs_01.qdcd_normalisation(dcd_norm_01_splev)

    # These interpolation methods could be decently different, but lets enforce
    # that our values are the same to within 1%.
    for i, test_intensity in enumerate(test_intensities):
        assert test_intensity == pytest.approx(scan2d_from_nxs_01.intensity[i],
                                               rel=0.01)

    for i, test_inten_e in enumerate(test_intensities_e):
        assert test_inten_e == pytest.approx(scan2d_from_nxs_01.intensity_e[i],
                                             rel=0.01)

def test_footprint_correction_01(scan2d_from_nxs_01: Scan2D):
    """
    Makes sure that the footprint correction acually does something for a
    reasonable beam FWHM and a small (1mm) sample.
    """
    # 100 micron beam.
    beam_width = 100e-6
    # 1 mm sample.
    sample_size = 1e-3
    intensities_0 = np.copy(scan2d_from_nxs_01.intensity)
    intensities_e_0 = np.copy(scan2d_from_nxs_01.intensity_e)
    scan2d_from_nxs_01.footprint_correction(beam_width, sample_size)

    assert (intensities_0 != scan2d_from_nxs_01.intensity).all()
    assert (intensities_e_0 != scan2d_from_nxs_01.intensity_e).all()

def test_footprint_correction_02(scan2d_from_nxs_01: Scan2D):
    """
    Do a really naive footprint correction assuming a step function beam.
    Enforce that this is the same as our fancy correction, to within 10%.
    (Note: they are actually about 10% out from each other).
    """

    # 100 micron beam.
    beam_width = 100e-6
    # 1 mm sample.
    sample_size = 1e-3

    intensities_0 = np.copy(scan2d_from_nxs_01.intensity)
    intensities_e_0 = np.copy(scan2d_from_nxs_01.intensity_e)

    beam_size_on_sample = beam_width / \
        np.sin(np.radians(scan2d_from_nxs_01.theta))

    incident_beam_fraction = sample_size / beam_size_on_sample

    test_intensities = intensities_0/incident_beam_fraction
    test_intensities_e = intensities_e_0/incident_beam_fraction

    scan2d_from_nxs_01.footprint_correction(beam_width, sample_size)
    for i, test_intensity in enumerate(test_intensities):
        assert test_intensity == pytest.approx(
            scan2d_from_nxs_01.intensity[i], 0.1)

    for i, test_intensity_e in enumerate(test_intensities_e):
        assert test_intensity_e == pytest.approx(
            scan2d_from_nxs_01.intensity_e[i], 0.1)

def test_crop_01(scan2d_from_nxs_01: Scan2D, region_01):
    """
    Check that crop is decreasing the size of the image.
    """
    initial_shape = scan2d_from_nxs_01.images[0].shape
    scan2d_from_nxs_01.crop(crop_to_region, region=region_01)

    assert scan2d_from_nxs_01.images[0].shape[0] < initial_shape[0]
    assert scan2d_from_nxs_01.images[0].shape[1] < initial_shape[1]

def test_crop_02(scan2d_from_nxs_01: Scan2D, region_01: Region):
    """
    Make sure that our cropped region has the correct size.
    """
    scan2d_from_nxs_01.crop(crop_to_region, region=region_01)
    assert (scan2d_from_nxs_01.images[0].shape[0]
            * scan2d_from_nxs_01.images[0].shape[1]) == region_01.num_pixels

def test_crop_03(scan2d_from_nxs_01: Scan2D, region_01: Region):
    """
    Make sure that the region we've cropped to has the specified shape.
    """
    scan2d_from_nxs_01.crop(crop_to_region, region=region_01)
    assert scan2d_from_nxs_01.images[0].shape[0] == region_01.x_length
    assert scan2d_from_nxs_01.images[0].shape[1] == region_01.y_length

def test_bkg_sub_01(scan2d_from_nxs_01: Scan2D):
    """
    Make sure that we start out with no background.
    """
    assert scan2d_from_nxs_01.images[0].bkg == 0
    assert scan2d_from_nxs_01.images[0].bkg_e == 0

def test_bkg_sub_02(scan2d_from_nxs_01: Scan2D):
    """
    Make sure that the background subtraction function is doing something.
    """
    region_list = scan2d_from_nxs_01.metadata.background_regions
    scan2d_from_nxs_01.bkg_sub(roi_subtraction, list_of_regions=region_list)

    assert scan2d_from_nxs_01.images[0].bkg != 0
    assert scan2d_from_nxs_01.images[0].bkg_e != 0

def test_bkg_sub_03(scan2d_from_nxs_01: Scan2D):
    """
    Make sure that the background subtraction decreases our intensity.
    """
    vals, stdevs = (np.zeros(
        len(scan2d_from_nxs_01.intensity)),
        np.zeros(len(scan2d_from_nxs_01.intensity)))

    # Also update the image intensities & errors.
    for i, image in enumerate(scan2d_from_nxs_01.images):
        vals[i], stdevs[i] = image.sum()

    # Store the intensity(Q) to the new value.
    scan2d_from_nxs_01.intensity = np.array(vals)
    scan2d_from_nxs_01.intensity_e = np.array(stdevs)

    region_list = scan2d_from_nxs_01.metadata.background_regions
    scan2d_from_nxs_01.bkg_sub(roi_subtraction, list_of_regions=region_list)

    assert (vals > scan2d_from_nxs_01.intensity).all()

def test_bkg_sub_04(scan2d_from_nxs_01: Scan2D, scan2d_from_nxs_01_copy,
                    custom_bkg_region_01):
    """
    Make sure that using two background regions yields a lower uncertainty
    measurement of the background than using just one background region.
    """
    regions_1 = [scan2d_from_nxs_01.metadata.background_regions[0]]
    regions_2 = [scan2d_from_nxs_01.metadata.background_regions[0]] + [
        custom_bkg_region_01]
    scan2d_from_nxs_01.bkg_sub(roi_subtraction, list_of_regions=regions_1)
    scan2d_from_nxs_01_copy.bkg_sub(
        roi_subtraction, list_of_regions=regions_2)

    for i, image_1 in enumerate(scan2d_from_nxs_01.images):
        image_2 = scan2d_from_nxs_01_copy.images[i]
        assert image_1.bkg_e > image_2.bkg_e

def test_gauss_bkg_01(scan2d_from_nxs_01: Scan2D):
    """
    Make sure that our Gaussian fit background subtraction function is doing
    something.

    Note that this function is not being tested for sensible results because
    this doesn't generally seem to be a sensible technique to use on I07. As
    more instruments are supported, if this technique becomes useful, its
    tests will need to be extended. For now, only the minimum is being done
    to ensure that it is roughly functional.
    """
    scan2d_from_nxs_01.bkg_sub(fit_gaussian_1d)

    assert scan2d_from_nxs_01.images[0].bkg != 0
    assert scan2d_from_nxs_01.images[0].bkg_e != 0

def test_gauss_bkg_02(scan2d_from_nxs_01: Scan2D):
    """
    Make sure that carrying out this subtraction decreases our intensity.

    Note that this function is not being tested for sensible results because
    this doesn't generally seem to be a sensible technique to use on I07. As
    more instruments are supported, if this technique becomes useful, its
    tests will need to be extended. For now, only the minimum is being done
    to ensure that it is roughly functional.
    """
    vals = np.zeros(len(scan2d_from_nxs_01.intensity))
    # Also update the image intensities & errors.
    for i, image in enumerate(scan2d_from_nxs_01.images):
        vals[i], _ = image.sum()
    # Store the intensity(Q) to the new value.
    scan2d_from_nxs_01.intensity = np.array(vals)

    intensity_0 = np.copy(scan2d_from_nxs_01.intensity)
    scan2d_from_nxs_01.bkg_sub(fit_gaussian_1d)

    assert (scan2d_from_nxs_01.intensity < intensity_0).all()

max_stars_repo_name: RBrearton/islatu

max_issues_repo_name: RBrearton/islatu

max_forks_repo_name: RBrearton/islatu