text:
# -*- coding: utf-8 -*-
# /tests/test_nbody.py

"""
----------------------------------------------------------------------------
tests for cheby_checker/nbody
 - Here I focus on the functions that RUN the nbody integrations.
 - In the accompanying test_nbody_parse script I test the functions for PARSING the input data

Jan 2022
Matthew Payne

Prev Work:
Mike Alexandersen, Matthew Payne & Matthew Holman

This code simplified as of Jan 2022
Removing many tests of non-json input
 - The non-json input methods *may* still work, but for now I just want to ensure that the json inputs work
 - Old tests of the non-json input can be found in the tests/archiaic/ttestt_nbody* file(s)
 tests for mpc_nbody

----------------------------------------------------------------------------
"""

# import third-party packages
# -----------------------------------------------------------------------------
import sys
import os
import numpy as np
import pytest
import time
import glob

# Import neighbouring packages
# -----------------------------------------------------------------------------
sys.path.append(os.environ['REBX_DIR'])
from examples.ephem_forces import ephem_forces

# The main nbody code we are trying to test
from cheby_checker import nbody

# old conversion library that may be useful for cross-comparison of various tests ...
from cheby_checker import MPC_library as mpc

import convenience_Horizons as Horizons

this_dir = os.path.abspath(os.path.dirname( __file__ ))
repo_dir = os.path.abspath(os.path.dirname( this_dir ))
data_dir = os.path.join(repo_dir, 'dev_data')
json_dir = os.path.join(data_dir, 'mpc_orb_jsons')
std_json_dir = os.path.join(json_dir, 'standard_mp') # Standard grav-only fits

# Utility functions to help with testing
# -----------------------------------------------------------------------------

def similar_xyzuvw(xyzv0, xyzv1, threshold_xyz=1e-13, threshold_v=1e-14): # 15 mm, 1.5 mm/day
    """
    Calculate the difference between two sets of cartesian coordinates.
    """
    if isinstance(xyzv0, list):
        xyzv0 = np.array(xyzv0)
    if isinstance(xyzv1, list):
        xyzv1 = np.array(xyzv1)
    error = xyzv0 - xyzv1
    good_tf = np.abs(error) < np.array([threshold_xyz] * 3 + [threshold_v] * 3)
    return np.all(good_tf), error

def is_nbody_output_good_enough(times, data, target='30102'):
    """
    Mike Alexandersen & Matt Payne
    Helper function for determining whether the saved output from an nbody
    integration is good enough.
    """
    # Check 20 timesteps
    # NB If len(times)<20, indicees will initially contain repeats, but we subsequently force uniqueness using "set"
    indicees = np.linspace(0, len(times) - 1, 20).astype(int)
    for i in set(indicees):

        # Get Horizons "state"" for that time
        horizons_xyzv = nice_Horizons(target, '500@0', times[i], 'smallbody')

        # The "state" from our integration
        mpc_xyzv = data[i, 0, :]

        # Check whether our integration agrees with Horizons ( within threshold )
        error, good_tf = compare_xyzv(horizons_xyzv,
                                        mpc_xyzv,
                                        1e-7, 1e-8) # MJP 2020-09-03 : Artificially increased thresholds to allow me to make progress while waiting for Holman to debug
                                        #5e-11, 2e-13)  # 7.5m, 30 mm/day
        if np.all(good_tf):
            pass # print('Awesome!')
        else:
            print(f'\n Discrepancy in *is_nbody_output_good_enough()* ...')
            print(f'Time, timestep: {times[i]:}, {i:}')
            print(f'Horizons : {["%18.15e" % _ for _ in horizons_xyzv]}')
            print(f'N-body   : {["%18.15e" % _ for _ in mpc_xyzv]}')
            print(f'Position off by [au]: {error[:3]:}')
            print(f'Velocity off by [au/day]: {error[3:6]:}')
        assert np.all(good_tf)

# Tests of NbodySim
# -----------------------------------------------------------------------------

def test_nbody_A():
    """
    Test instantiation of NbodySim object
    """
    # Instantiate
    N = nbody.NbodySim()

    # Check that the expected attributes exist
    assert \
        N.tstart             == None  and \
        N.tstop              == None  and \
        N.geocentric         == False  and \
        N.integration_epoch  == None  and \
        N.mpcorb_list        == []  and \
        N.input_n_particles  == None  and \
        N.save_output        == False  and \
        N.save_output_file   == None  and \
        N.verbose            == False  and \
        N.CHECK_EPOCHS       == True  and \
        N.unpacked_primary_provisional_designation_list == [] and \
        N.helio_ecl_vec_EXISTS   == False  and \
        N.helio_ecl_vec          == None  and \
        N.helio_ecl_cov_EXISTS   == False  and \
        N.helio_ecl_cov          == None  and \
        N.bary_eq_vec_EXISTS     == False  and \
        N.bary_eq_vec            == None  and \
        N.bary_eq_cov_EXISTS     == False  and \
        N.bary_eq_cov            == None  and \
        N.non_grav_EXISTS        == False  and \
        N.non_grav_dict_list     == []  and \
        N.output_times       == None  and \
        N.output_states      == None  and \
        N.output_covar       == None

def test_integration_function_A():
    """
    If we put ANYTHING into the ephem_forces.integration_function,
    will it work or crash and burn?

    Most likely if there is a problem, it'll cause pytest to crash entirely,
    so might as well start with this.

    Note that *ephem_forces.integration_function* is NOT the main function that will
    be called by cheby_checker.nbody, but it IS called by
    *ephem_forces.production_integration_function_wrapper*, which IS called by
    cheby_checker.nbody

    Note that this does NOT perform any tests of integration accuracy:
    I am purely trying to get a simple call to execute

    """

    # Define the initial positions. We use the state for Asteroid (3666) Holman from DE441
    # 2458849.500000000 = A.D. 2020-Jan-01 00:00:00.0000 TDB [del_T=     69.183900 s]
    # X = 3.338875350265349E+00 Y =-9.176518267602161E-01 Z =-5.038590677470149E-01
    # VX= 2.805663315227095E-03 VY= 7.550408688437705E-03 VZ= 2.980028207454247E-03
    tstart = 2458849.5
    instates = np.array([[3.338875349745594E+00, -9.176518281675284E-01, -5.038590682977396E-01, 2.805663319000732E-03, 7.550408687780768E-03, 2.980028206579994E-03]])

    # Define the length of the integration (days)
    tstep = 20.0
    tend = tstart + 1.0

    # Define the quantities associated with the variational equaitons
    n_var = 6
    n_particles = 7
    geocentric = 0
    invar_part = np.zeros(6, dtype=int)
    invar = np.identity(6)

    # Run the integration
    times, states, var, var_ng, status = ephem_forces.integration_function(
        tstart, tend, tstep, geocentric, n_particles, instates, n_var, invar_part, invar)

    assert isinstance(states, np.ndarray)
    assert isinstance(times, np.ndarray)
    assert isinstance(var, np.ndarray)

def test_integration_function_B():
    """
    ... ephem_forces.integration_function,
    ...

    """

    # Define the variables that will be used in the query
    target  = '12345'
    centre  = '500@0'
    epochs  = ['2458850.0','2458880.0']
    id_type = 'smallbody'
    refplane='earth'

    # Call the *nice_Horizons* function to get the cartesian states at the first time
    # This is returning EQUATORIAL BERYCENTRIC coordinates
    horizons_zero = Horizons.nice_Horizons(target, centre, epochs[0], id_type, refplane=refplane)

    # Run the integration
    tstart = float(epochs[0])
    tend   = float(epochs[-1])
    tstep  = 0.01
    geocentric = 0
    n_particles = 1
    instates = np.array([horizons_zero])
    n_var = 6
    invar_part = np.zeros(6, dtype=int)
    invar = np.identity(6)
    times, states, var, var_ng, status = ephem_forces.integration_function(tstart,
                                                                            tend,
                                                                            tstep,
                                                                            geocentric,
                                                                            n_particles,
                                                                            instates,
                                                                            n_var,
                                                                            invar_part,
                                                                            invar)

    # Now call horizons again at some of the output times at which the integration_function produced output
    for n, t in enumerate(times):

        # Check horizons at the simulation-time output
        h = Horizons.nice_Horizons(target, centre, t, id_type, refplane=refplane)

        # Check similarity: threshold_xyz=1e-11, threshold_v=1e-12 : # 150 cm, 15 cm/day
        similar_bool , error = similar_xyzuvw(h, states[n][0], threshold_xyz=1e-11, threshold_v=1e-12)
        assert similar_bool

def test_production_integration_function_wrapper_A():
    """
    First test of *ephem_forces.production_integration_function_wrapper*
    Doing a 2-particle integration
    Testing the STRUCTURE of the returned arrays
    *NOT* testing the numerical accuracy

    """
    tstart  = 2456184.7
    tstop   = tstart + 600
    epoch   = tstart

    geocentric = 0
    n_particles = 2
    instates = np.array([
                        [-3.1, 2.7, 3.6, -0.006, -0.004, -0.002] ,
                        [-4.1, 3.7, 5.6, -0.004, -0.003, -0.002]
                        ])

    # Call the function
    outtimes, states, partial_derivatives_wrt_state, partial_derivatives_wrt_NG, return_value = ephem_forces.production_integration_function_wrapper(  tstart,
                                                                tstop,
                                                                epoch,
                                                                instates,
                                                                non_grav_dict_list = None,
                                                                geocentric = 0 ,
                                                                epsilon = 1e-8,
                                                                tstep_min = 0.02,
                                                                tstep_max = 32.)
    # check that we get the expected variable-types back
    assert isinstance(states, np.ndarray)
    assert isinstance(outtimes, np.ndarray)
    assert isinstance(partial_derivatives_wrt_state, np.ndarray)
    assert isinstance(return_value, tuple)

    # check that the SHAPES of the returned arrays are as expected ...
    # outtimes.shape                        ~ (161,)
    # states.shape                          ~ (161, 2, 6)
    # partial_derivatives_wrt_state.shape   ~ (161, 12, 6)
    assert  outtimes.shape[0] == states.shape[0] == partial_derivatives_wrt_state.shape[0]
    assert  states.ndim ==3 and \
            states.shape[1] == n_particles and \
            states.shape[2] == 6
    assert  partial_derivatives_wrt_state.ndim ==3 and \
            partial_derivatives_wrt_state.shape[1] == 6*n_particles and \
            partial_derivatives_wrt_state.shape[2] == 6

def test_production_integration_function_wrapper_B():
    """
    Doing a timing test of the production_integration_function_wrapper

    When running directly from a file in ephem_forcs, this test takes
    ~1 secs to run (20,000 day integration)

    During development there were issues in which the same query run from a
    different directory would take ~3mins (i.e. 100 times longer)

    I want to ensure that we don't have this problem
    """
    start_time = time.time()
    #print(__file__, ':test_production_integration_function_wrapper_B')

    # Define the inputs
    instates = np.array([[3.338875349745594E+00, -9.176518281675284E-01, -5.038590682977396E-01, 2.805663319000732E-03, 7.550408687780768E-03, 2.980028206579994E-03]])
    tstart, trange = 2458849.5, 20000
    epoch = tstart
    tend = tstart + trange

    # Run the integration
    times, states, var, var_ng, status = ephem_forces.production_integration_function_wrapper(tstart, tend, epoch, instates)

    # Check that the execution time is less than 5 seconds (when it gets bad it ballons to 3 minutes)
    end_time = time.time()
    allowed_time = 5 #seconds
    print('test_production_integration_function_wrapper_B: end_time-start_time',end_time-start_time)
    assert end_time-start_time < allowed_time

def test_production_integration_function_wrapper_C():
    """
    Another test of timing...
    In practice 10 particles seemed to take ~2.5 secs (end-start below)
    """
    start_time = time.time()
    #print(__file__, ':test_production_integration_function_wrapper_B')

    # Define the inputs
    instates = np.array([
        [3.338875349745594E+00, -9.176518281675284E-01, -5.038590682977396E-01, 2.805663319000732E-03, 7.550408687780768E-03, 2.980028206579994E-03],
        [3.388875349745594E+00, -9.176518281675284E-01, -5.038590682977396E-01, 2.805663319000732E-03, 7.540408687780768E-03, 2.980028206579994E-03],
        [3.438875349745594E+00, -9.176518281675284E-01, -5.038590682977396E-01, 2.805663319000732E-03, 7.530408687780768E-03, 2.980028206579994E-03],
        [3.488875349745594E+00, -9.176518281675284E-01, -5.038590682977396E-01, 2.805663319000732E-03, 7.520408687780768E-03, 2.980028206579994E-03],
        [3.538875349745594E+00, -9.176518281675284E-01, -5.038590682977396E-01, 2.805663319000732E-03, 7.510408687780768E-03, 2.980028206579994E-03],
        [3.588875349745594E+00, -9.176518281675284E-01, -5.038590682977396E-01, 2.805663319000732E-03, 7.550408687780768E-03, 2.980028206579994E-03],
        [3.638875349745594E+00, -9.176518281675284E-01, -5.038590682977396E-01, 2.805663319000732E-03, 7.540408687780768E-03, 2.980028206579994E-03],
        [3.688875349745594E+00, -9.176518281675284E-01, -5.038590682977396E-01, 2.805663319000732E-03, 7.530408687780768E-03, 2.980028206579994E-03],
        [3.738875349745594E+00, -9.176518281675284E-01, -5.038590682977396E-01, 2.805663319000732E-03, 7.520408687780768E-03, 2.980028206579994E-03],
        [3.788875349745594E+00, -9.176518281675284E-01, -5.038590682977396E-01, 2.805663319000732E-03, 7.510408687780768E-03, 2.980028206579994E-03],
    ])
    tstart, trange = 2458849.5, 20000
    epoch = tstart
    tend = tstart + trange

    # Run the integration
    times, states, var, var_ng, status = ephem_forces.production_integration_function_wrapper(tstart, tend, epoch, instates)

    # Check that the execution time is less than 15 seconds (when it gets bad it ballons to 3 minutes)
    end_time = time.time()
    allowed_time = 15 #seconds
    print('test_production_integration_function_wrapper_C: end_time-start_time',end_time-start_time)
    assert end_time-start_time < allowed_time

def test_production_integration_function_wrapper_D():
    """

    Testing the numerical ACCURACY of the results returned from a
    single run of the *ephem_forces.production_integration_function_wrapper*
    function.

    Note that many more tests of the accuracy of the reboundx integrator
    need to be performed, but they need to be done in the REBOUNDX
    package itself
    """

    # Define the variables that will be used in the query
    target  = '719' # Asteroid #123456
    centre  = '500@0'  # <<-- Barycentric
    epochs  = ['2458000.0','2458200.0']
    id_type = 'smallbody'
    refplane='earth' # <<--Equatorial

    # Call the *nice_Horizons* function to get the cartesian states at the first time
    # This is returning EQUATORIAL BERYCENTRIC coordinates
    horizons_zero = Horizons.nice_Horizons(target, centre, epochs[0], id_type, refplane=refplane)
    print('horizons_zero=',horizons_zero)

    # Call the production_integration_function_wrapper that we want to test
    # I believe that the integration is performed in EQUATORIAL BERYCENTRIC coordinates
    tstart = epoch = epochs[0]
    tstop  = epochs[-1]
    instates = np.array([ horizons_zero ])
    outtimes, states, partial_derivatives_wrt_state, partial_derivatives_wrt_NG, return_value = ephem_forces.production_integration_function_wrapper(   float(tstart),
                                                                float(tstop),
                                                                float(epoch),
                                                                instates,
                                                                non_grav_dict_list = None,
                                                                geocentric = 0 ,
                                                                epsilon = 1e-10,
                                                                tstep_min = 0.02,
                                                                tstep_max = 32.)

    # Now call horizons again at some of the output times at which the *production_integration_function_wrapper()* produced output
    for n, t in enumerate(outtimes):

        # Check horizons at the simulation-time output
        h = Horizons.nice_Horizons(target, centre, t, id_type, refplane=refplane)

        # Check similarity: threshold_xyz=1e-11, threshold_v=1e-13 : # 15 m, 1.5 m/day
        similar_bool , error = similar_xyzuvw(h, states[n][0], threshold_xyz=1e-10, threshold_v=1e-11)
        assert similar_bool, f'test_production_integration_function_wrapper_D:n={n}, error={error}'

def test_run_integration_A():
    """
    Test the NbodySim.__run_integration() function
    This
        calls *production_integration_function_wrapper*
        then performs various functions to reshape the partial derivs,
        then calculates the cov-matrix at each timestep

    We want to check that the output has the expected type & shape
    """

    # Instantiate
    N = nbody.NbodySim()

    # Declare inputs
    tstart = 2459200
    tstop  = 2459300
    data_file = '2000SR210.json'
    mpcorb_list = [ os.path.join(std_json_dir , data_file) ]

    # Parse the inputs
    N._parse_inputs_run_mpcorb(tstart = tstart , tstop = tstop , mpcorb_list = mpcorb_list)

    # Convert to barycentric equatorial
    N.make_bary_equatorial()

    # Check some quantities BEFORE running the test function ...
    assert N.input_n_particles == 1
    assert N.output_times   == None
    assert N.output_states  == None
    assert N.output_covar   == None

    # ## ### #### ##### ######
    # Run the *__run_integration* function we want to test
    # ## ### #### ##### ######
    N.verbose = True
    N._run_integration()

    # Check the output has the expected type & shape
    assert isinstance(N.output_times, np.ndarray)
    assert isinstance(N.output_states, np.ndarray)
    assert isinstance(N.output_covar, np.ndarray)

    # output_times.shape   ~ (161,)
    # output_states.shape  ~ (161, 1, 6)
    # output_covar.shape   ~ (161, 1, 6, 6)
    n_particles = 1
    assert  N.output_times.shape[0] == N.output_states.shape[0] == N.output_covar.shape[0]
    assert  N.output_states.ndim ==3 and \
            N.output_states.shape[1] == n_particles and \
            N.output_states.shape[2] == 6
    assert  N.output_covar.ndim ==4 and \
            N.output_covar.shape[1] == n_particles and \
            N.output_covar.shape[2] == 6 and \
            N.output_covar.shape[3] == 6

def test_run_mpcorb_A():
    """
    Test the overall *run_mpcorb* function

    Note that here we are testing that
    (a) the cartesian positions from the orbfit fit are "close enough" to the JPL fit (10's of km)
    (b) the cartesian positions remain close to the JPL fit throughout the integration by reboundx
    """

    # Instantiate
    N = nbody.NbodySim()

    # Declare inputs
    tstart = 2459200
    tstop  = 2459300
    data_file = '2000SR210.json'
    mpcorb_list = [ os.path.join(std_json_dir , data_file) ]

    # Run the *run_mpcorb* that we want to test
    N.run_mpcorb( tstart = tstart , tstop = tstop , mpcorb_list = mpcorb_list )

    # Define the variables that will be used in the query
    target  = '56986' # 2000SR210 == Asteroid #56986
    centre  = '500@0'  # <<-- Barycentric
    id_type = 'smallbody'
    refplane='earth' # <<--Equatorial

    # Now call horizons again at the output times at which the *production_integration_function_wrapper()* produced output
    for n, t in enumerate(N.output_times):

        # Check horizons at the simulation-time output
        h = Horizons.nice_Horizons(target, centre, t, id_type, refplane=refplane)

        # Check similarity: threshold_xyz=1e-6, threshold_v=1e-8
        # The expected differences are at the few-times 10^-7 level => few-times 10km
        #  -- This seems entirely reasonable for the differences in the results of orbit fits
        similar_bool , error = similar_xyzuvw(h, N.output_states[n][0], threshold_xyz=1e-6, threshold_v=1e-8)
        assert similar_bool

def test_run_mpcorb_B():
    """
    Test the overall *run_mpcorb* function on MULTIPLE INPUT FILES

    """

    # Instantiate
    N = nbody.NbodySim()

    # Declare inputs
    tstart = 2459200
    tstop  = 2459300
    mpc_orb_json_files = glob.glob(std_json_dir + '/*.json' )[:3]

    # Run the *run_mpcorb* that we want to test
    N.run_mpcorb( tstart = tstart , tstop = tstop , mpcorb_list = mpc_orb_json_files )

    # Add some tests !!!


text:
import os
import sys

sys.path.append(
    os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir))
)

import numpy as np
import oneflow as flow
import oneflow._oneflow_internal

from oneflow_gpt.config import get_args
from oneflow_gpt import distribute as dist
from oneflow_gpt.data import GPTDataLoader
from oneflow_gpt.model import GPTModel, Embedding, Logits
from oneflow_gpt.model import Transformer, TransformerLayer, ActivationCheckpointing
from oneflow_gpt.model import ParallelSparseSoftmaxCrossEntropyLoss
from oneflow_gpt.optimizer import make_optimizer, make_lr_scheduler, make_grad_scaler
from oneflow_gpt.logger import print_rank_0, print_rank_last, Logger
from oneflow_gpt.logger import IterationMetric, AccumulationMetric
from oneflow_gpt.logger import LossMetric, ThroughputMetric

class Trainer(object):
    def __init__(self):
        self.args = get_args()
        self.rank = flow.env.get_rank()
        self.world_size = flow.env.get_world_size()
        self.model = GPTModel()
        self.data_loader = GPTDataLoader()
        self.cross_entropy = ParallelSparseSoftmaxCrossEntropyLoss()
        self.optimizer = make_optimizer(self.args, self.model)
        self.lr_scheduler = make_lr_scheduler(self.args, self.optimizer)
        # self.optimizer = None
        # self.lr_scheduler = None
        # NOTE(zwx): grad scaler is not available in eager mode
        self.grad_scaler = make_grad_scaler(self.args)

        if self.args.graph:
            flow.boxing.nccl.enable_use_compute_stream(True)

            self.train_graph = GPTGraph(
                self.model,
                self.data_loader,
                self.cross_entropy,
                self.optimizer,
                self.lr_scheduler,
                self.grad_scaler,
            )

        # self.save("init")

        self.logger = Logger(self.rank)
        self.logger.register_metric("iter", IterationMetric())
        self.logger.register_metric("samples", AccumulationMetric())
        self.logger.register_metric("loss", LossMetric(), "loss: {:.5f}", True)
        self.logger.register_metric(
            "throughput", ThroughputMetric(), "throughput: {:.2f}", True
        )

    def __call__(self):
        iteration = 0
        while iteration < self.args.train_iters:
            if self.args.graph:
                loss = self.train_graph()
            else:
                raise NotImplementedError
                # loss = self.train_eager()

            if loss.is_consistent:
                loss = loss.to_local()

            # snapshot.step()
            # iteration = snapshot.iter
            iteration += 1
            print(f"[{self.rank}] iteration: {iteration}")

            self.logger.meter("samples", self.args.global_batch_size)
            self.logger.meter("loss", loss)
            self.logger.meter("throughput", self.args.global_batch_size)
            if iteration % self.args.log_interval == 0:
                self.logger.meter("iter", iteration)
                self.logger.print_metrics([self.world_size - 1])

        print(f"[{self.rank}] training finished")

    def train_eager(self):
        data, label = self.data_loader()
        logits = self.model(data)
        loss = self.cross_entropy(logits, label)
        loss.backward()
        self.optimizer.zero_grad()
        self.optimizer.step()
        self.lr_scheduler.step()
        return loss

    def save(self, subdir):
        if self.args.checkpoint_save_path is None:
            return

        save_path = os.path.join(self.args.checkpoint_save_path, subdir)
        print_rank_0(f"Saving model to {save_path}")
        state_dict = self.model.state_dict()

        flow.save(state_dict, save_path, consistent_dst_rank=0)

class GPTGraph(flow.nn.Graph):
    def __init__(
        self,
        model,
        data_loader,
        cross_entropy,
        optimizer=None,
        lr_scheduler=None,
        grad_scaler=None,
    ):
        super().__init__()
        self.model = model
        self.data_loader = data_loader
        self.cross_entropy = cross_entropy
        self.is_train = False
        if optimizer is not None:
            self.is_train = True
            self.add_optimizer(optimizer, lr_sch=lr_scheduler)
            if grad_scaler is not None:
                self.set_grad_scaler(grad_scaler)

        args = get_args()
        self.set_activation_checkpointing()
        self.set_pipeline_stage_id()
        self.config.set_gradient_accumulation_steps(args.num_accumulation_steps)

        if args.fp16:
            self.config.enable_amp(True)

        self.config.allow_fuse_add_to_output(True)
        self.config.allow_fuse_model_update_ops(True)
        self.config.allow_fuse_cast_scale(True)

    def set_activation_checkpointing(self):
        for module_block in self.model.modules():
            if isinstance(module_block.origin, TransformerLayer):
                module_block.config.activation_checkpointing = True

    def set_pipeline_stage_id(self):
        dist_util = dist.get_dist_util()

        self.data_loader.config.stage_id = dist_util.get_layer_stage_id(0)
        self.data_loader.data_decoder.config.stage_id = dist_util.get_layer_stage_id(0)

        for module_block in self.model.modules():
            if isinstance(module_block.origin, Embedding):
                module_block.config.stage_id = dist_util.get_layer_stage_id(0)
            elif isinstance(
                module_block.origin, (TransformerLayer, ActivationCheckpointing)
            ):
                module_block.config.stage_id = dist_util.get_layer_stage_id(
                    module_block.origin.layer_idx
                )
            elif isinstance(module_block.origin, Transformer):
                module_block.config.stage_id = dist_util.get_layer_stage_id(-1)
            elif isinstance(module_block.origin, Logits):
                module_block.config.stage_id = dist_util.get_layer_stage_id(-1)
            else:
                pass

        self.data_loader.label_decoder.config.stage_id = dist_util.get_layer_stage_id(
            -1
        )
        self.cross_entropy.config.stage_id = dist_util.get_layer_stage_id(-1)

    def build(self):
        data, label = self.data_loader()
        logits = self.model(data)
        loss = self.cross_entropy(logits, label)
        if self.is_train:
            loss.backward()
        return loss

if __name__ == "__main__":
    Trainer()()
    print(f"[{flow.env.get_rank()}] exit")

ext: py

reason: import numpy


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

# Function to generate corrected axon+myelin image after correcting the myelin mask.

import os
import numpy as np

# AxonDeepSeg imports
import AxonDeepSeg.ads_utils as ads
from AxonDeepSeg.ads_utils import convert_path

def generate_axons_from_myelin(path_prediction,path_myelin_corrected):
    """
    :param path_prediction: path of the prediction i.e. image of axon+myelin segmentation (output of AxonDeepSeg)
    :param path_myelin_corrected: path of corrected myelin by the user i.e. myelin mask (uint8 type with myelin=255, background=0)
    :return: merged and corrected axon+myelin image
    """

    # If string, convert to Path objects
    path_prediction = convert_path(path_prediction)
    path_myelin_corrected = convert_path(path_myelin_corrected)

    # read output from axondeepseg and myelin mask corrected by user
    prediction = ads.imread(path_prediction)
    myelin_corrected = ads.imread(path_myelin_corrected)

    # compute the axon mask from axondeepseg (axon=255, myelin=127, background=0)
    axon_ads = prediction > 200

    # get the myelin mask corrected by user (myelin=255, background=0)
    myelin_corrected = myelin_corrected > 200

    # compute logical OR between axondeepseg axon mask and myelin corrected mask
    fused = np.logical_or(axon_ads, myelin_corrected)

    # compute new axon mask by logical XOR between corrected myelin mask and fused
    new_axon_mask = np.logical_xor(myelin_corrected, fused)

    # merge corrected myelin mask and generated axon mask
    both = new_axon_mask*255 + myelin_corrected*127

    # save the corrected axon+myelin image
    path = path_prediction.parent / 'axon_myelin_mask_corrected.png'
    ads.imwrite(path, both)

    return both

max_stars_repo_name: wukathryn/axondeepseg

max_issues_repo_name: wukathryn/axondeepseg

max_forks_repo_name: wukathryn/axondeepseg


text:
import numpy as np
import sys

from .audiobuffer import AudioBuffer
from .audioprocess import ProcessUnit
from .audioencoder import WavEncoder, write_wav

def _print_progress(iteration, total, prefix="", suffix="", decimals=1, barLength=50):
    """Represent the progress state through percentage on the terminal.
    """
    formatStr = "{0:." + str(decimals) + "f}"
    percent = formatStr.format(100 * (iteration / float(total)))
    filledLength = int(round(barLength * iteration / float(total)))
    bar = "#" * filledLength + "-" * (barLength - filledLength)
    sys.stdout.write("\r%s |%s| %s%s %s" % (prefix, bar, percent, "%", suffix)),
    if iteration == total:
        sys.stdout.write("\n")
    sys.stdout.flush()

class AudioProcess(object):
    """Main audio process
        This class interacts with other applications, audio process unit, and audio buffer.

        Parameters
        ----------
        sample_rate (int): The sample rate of the audio data
        frame_size (int): The size of the frame when processing the data
        channels (int): The number of channels of the audio data
        zero_pad (bool): Whether to zero pad the audio data
        audiobuffer (AudioBuffer): The audio buffer
        processunit (ProcessUnit): The process unit
        type_stream (str): The type of stream to process('wav' or '[TODO] mic')
        in_file_path (str): The file path of the input file

        Notes
        -----
        This contains ONLY wav file processing. It will contains [TODO] Mic stream, real-time wave file stream,
        recording the processeed mic stream. Then, the type_stream can be changed to 'wav' or 'mic', which is currently
        only supported to 'wav'.

        TODO LIST
        ---------
        - Mic stream
        - Wave file stream
        - Recording the processed mic stream
    """

    def __init__(
        self,
        sample_rate,
        frame_size,
        channels,
        zero_pad=False,
        process_time_domain=None,
        process_fft_domain=None,
        process_post_domain=None,
        in_file_path="",
    ) -> None:
        self.sample_rate = sample_rate
        self.frame_size = frame_size
        self.channels = channels
        self.zero_pad = zero_pad

        self.audiobuffer = AudioBuffer(frame_size)
        self.processunit = ProcessUnit(
            sample_rate=sample_rate,
            frame_size=frame_size,
            zero_pad=zero_pad,
            process_time_domain=process_time_domain,
            process_fft_domain=process_fft_domain,
            process_post_domain=process_post_domain,
        )

        self.type_stream = "wav"
        self.in_file_path = in_file_path

    def process(self, indata, outdata, frames, time, status):
        """This is called (from a separate thread) for each audio block."""
        if status:
            print(status, file=sys.stderr)

        # Process the input data
        outdata[:] = self.processunit.stream_process(indata.copy())

        self.audiobuffer.inbuffer = indata
        self.audiobuffer.outbuffer = outdata

    def stream(self):
        if self.type_stream == "wav":
            NotImplementedError
        elif self.type_stream == "mic":
            NotImplementedError
        else:
            raise ValueError("type_stream must be 'wav' or 'mic'")

    def save(self, out_file_path="", progress_bar=True):
        if self.in_file_path == "":
            raise ValueError("in_file_path must be set")

        if self.type_stream == "wav":
            sample_rate = self.sample_rate
            callback = self.process
            frame_size = self.frame_size

            encoder = WavEncoder(self.in_file_path)
            in_data = encoder.get(frame_size)
            del encoder

            out_data = np.zeros_like(in_data)
            num_frames = in_data.shape[0] // frame_size

            for iframe in range(num_frames):
                in_frame = in_data[iframe * frame_size : (iframe + 1) * frame_size, :]
                out_frame = out_data[iframe * frame_size : (iframe + 1) * frame_size, :]
                callback(in_frame, out_frame, frames=None, time=None, status=None)
                out_data[iframe * frame_size : (iframe + 1) * frame_size, :] = out_frame

                if progress_bar:
                    _print_progress(iframe, num_frames)

            if out_file_path == "":
                return out_data
            else:
                write_wav(out_file_path, sample_rate, out_data)

        elif self.type_stream == "mic":
            NotImplementedError
        else:
            raise ValueError("type_stream must be 'wav' or 'mic'")

max_stars_repo_name: ooshyun/filterdesign

max_issues_repo_name: ooshyun/FilterDesign

max_forks_repo_name: ooshyun/FilterDesign


text:
import abc
import pickle
import time
from collections import OrderedDict
from copy import deepcopy

import gtimer as gt
import numpy as np

from rlkit.core import logger, eval_util
from rlkit.data_management.env_replay_buffer import EnvReplayBuffer
from rlkit.data_management.path_builder import PathBuilder
from rlkit.policies.base import ExplorationPolicy
from rlkit.torch.sac.policies import MakeDeterministic
from rlkit.samplers import PathSampler
from rlkit.envs.wrapped_absorbing_env import WrappedAbsorbingEnv

from gym.spaces import Dict

class BaseAlgorithm(metaclass=abc.ABCMeta):
    """
    base algorithm for single task setting
    can be used for RL or Learning from Demonstrations
    """
    def __init__(
            self,
            env,
            exploration_policy: ExplorationPolicy,
            training_env=None,
            eval_policy=None,
            eval_sampler=None,

            num_epochs=100,
            num_steps_per_epoch=10000,
            num_steps_between_train_calls=1000,
            num_steps_per_eval=1000,
            max_path_length=1000,
            min_steps_before_training=0,

            replay_buffer=None,
            replay_buffer_size=10000,

            freq_saving=1,
            save_replay_buffer=False,
            save_environment=False,
            save_algorithm=False,

            save_best=False,
            save_best_starting_from_epoch=0,
            best_key='AverageReturn', # higher is better

            no_terminal=False,
            wrap_absorbing=False,

            render=False,
            render_kwargs={},

            freq_log_visuals=1,

            eval_deterministic=False
        ):
        self.env = env
        self.training_env = training_env or pickle.loads(pickle.dumps(env))
        self.exploration_policy = exploration_policy

        self.num_epochs = num_epochs
        self.num_env_steps_per_epoch = num_steps_per_epoch
        self.num_steps_between_train_calls = num_steps_between_train_calls
        self.num_steps_per_eval = num_steps_per_eval
        self.max_path_length = max_path_length
        self.min_steps_before_training = min_steps_before_training

        self.render = render

        self.save_replay_buffer = save_replay_buffer
        self.save_algorithm = save_algorithm
        self.save_environment = save_environment
        self.save_best = save_best
        self.save_best_starting_from_epoch = save_best_starting_from_epoch
        self.best_key = best_key
        self.best_statistic_so_far = float('-Inf')

        if eval_sampler is None:
            if eval_policy is None:
                eval_policy = exploration_policy
            eval_policy = MakeDeterministic(eval_policy)
            eval_sampler = PathSampler(
                env,
                eval_policy,
                num_steps_per_eval,
                max_path_length,
                no_terminal=no_terminal,
                render=render,
                render_kwargs=render_kwargs
            )
        self.eval_policy = eval_policy
        self.eval_sampler = eval_sampler

        self.action_space = env.action_space
        self.obs_space = env.observation_space
        self.replay_buffer_size = replay_buffer_size
        if replay_buffer is None:
            assert max_path_length < replay_buffer_size
            replay_buffer = EnvReplayBuffer(
                self.replay_buffer_size,
                self.env,
                random_seed=np.random.randint(10000)
            )
        else:
            assert max_path_length < replay_buffer._max_replay_buffer_size
        self.replay_buffer = replay_buffer

        self._n_env_steps_total = 0
        self._n_train_steps_total = 0
        self._n_rollouts_total = 0
        self._do_train_time = 0
        self._epoch_start_time = None
        self._algo_start_time = None
        self._old_table_keys = None
        self._current_path_builder = PathBuilder()
        self._exploration_paths = []

        if wrap_absorbing:
            # needs to be properly handled both here and in replay buffer
            raise NotImplementedError()
        self.wrap_absorbing = wrap_absorbing
        self.freq_saving = freq_saving
        self.no_terminal = no_terminal

        self.eval_statistics = None
        self.freq_log_visuals = freq_log_visuals

    def train(self, start_epoch=0):
        self.pretrain()
        if start_epoch == 0:
            params = self.get_epoch_snapshot(-1)
            logger.save_itr_params(-1, params)
        self.training_mode(False)
        self._n_env_steps_total = start_epoch * self.num_env_steps_per_epoch
        gt.reset()
        gt.set_def_unique(False)
        self.start_training(start_epoch=start_epoch)

    def pretrain(self):
        """
        Do anything before the main training phase.
        """
        pass

    def start_training(self, start_epoch=0):
        self._current_path_builder = PathBuilder()
        observation = self._start_new_rollout()

        for epoch in gt.timed_for(
            range(start_epoch, self.num_epochs),
            save_itrs=True,
        ):
            self._start_epoch(epoch)
            for steps_this_epoch in range(self.num_env_steps_per_epoch):
                action, agent_info = self._get_action_and_info(observation)
                if self.render: self.training_env.render()

                next_ob, raw_reward, terminal, env_info = (
                    self.training_env.step(action)
                )
                if self.no_terminal: terminal = False
                self._n_env_steps_total += 1

                reward = np.array([raw_reward])
                terminal = np.array([terminal])
                self._handle_step(
                    observation,
                    action,
                    reward,
                    next_ob,
                    np.array([False]) if self.no_terminal else terminal,
                    absorbing=np.array([0., 0.]),
                    agent_info=agent_info,
                    env_info=env_info,
                )
                if terminal[0]:
                    if self.wrap_absorbing:
                        raise NotImplementedError()
                        '''
                        If we wrap absorbing states, two additional
                        transitions must be added: (s_T, s_abs) and
                        (s_abs, s_abs). In Disc Actor Critic paper
                        they make s_abs be a vector of 0s with last
                        dim set to 1. Here we are going to add the following:
                        ([next_ob,0], random_action, [next_ob, 1]) and
                        ([next_ob,1], random_action, [next_ob, 1])
                        This way we can handle varying types of terminal states.
                        '''
                        # next_ob is the absorbing state
                        # for now just taking the previous action
                        self._handle_step(
                            next_ob,
                            action,
                            # env.action_space.sample(),
                            # the reward doesn't matter
                            reward,
                            next_ob,
                            np.array([False]),
                            absorbing=np.array([0.0, 1.0]),
                            agent_info=agent_info,
                            env_info=env_info
                        )
                        self._handle_step(
                            next_ob,
                            action,
                            # env.action_space.sample(),
                            # the reward doesn't matter
                            reward,
                            next_ob,
                            np.array([False]),
                            absorbing=np.array([1.0, 1.0]),
                            agent_info=agent_info,
                            env_info=env_info
                        )
                    self._handle_rollout_ending()
                    observation = self._start_new_rollout()
                elif len(self._current_path_builder) >= self.max_path_length:
                    self._handle_rollout_ending()
                    observation = self._start_new_rollout()
                else:
                    observation = next_ob

                if self._n_env_steps_total % self.num_steps_between_train_calls == 0:
                    gt.stamp('sample')
                    self._try_to_train(epoch)
                    gt.stamp('train')

            gt.stamp('sample')
            self._try_to_eval(epoch)
            gt.stamp('eval')
            self._end_epoch()

    def _try_to_train(self, epoch):
        if self._can_train():
            self.training_mode(True)
            self._do_training(epoch)
            self._n_train_steps_total += 1
            self.training_mode(False)

    def _try_to_eval(self, epoch):

        if self._can_evaluate():
            # save if it's time to save
            if (epoch % self.freq_saving == 0) or (epoch + 1 >= self.num_epochs):
                # if epoch + 1 >= self.num_epochs:
                #     epoch = 'final'
                logger.save_extra_data(self.get_extra_data_to_save(epoch))
                params = self.get_epoch_snapshot(epoch)
                logger.save_itr_params(epoch, params)

            self.evaluate(epoch)

            logger.record_tabular(
                "Number of train calls total",
                self._n_train_steps_total,
            )
            logger.record_tabular(
                "Number of env steps total",
                self._n_env_steps_total,
            )
            logger.record_tabular(
                "Number of rollouts total",
                self._n_rollouts_total,
            )

            times_itrs = gt.get_times().stamps.itrs
            train_time = times_itrs['train'][-1]
            sample_time = times_itrs['sample'][-1]
            eval_time = times_itrs['eval'][-1] if epoch > 0 else 0
            epoch_time = train_time + sample_time + eval_time
            total_time = gt.get_times().total

            logger.record_tabular('Train Time (s)', train_time)
            logger.record_tabular('(Previous) Eval Time (s)', eval_time)
            logger.record_tabular('Sample Time (s)', sample_time)
            logger.record_tabular('Epoch Time (s)', epoch_time)
            logger.record_tabular('Total Train Time (s)', total_time)

            logger.record_tabular("Epoch", epoch)
            logger.dump_tabular(with_prefix=False, with_timestamp=False)
        else:
            logger.log("Skipping eval for now.")

    def _can_evaluate(self):
        """
        One annoying thing about the logger table is that the keys at each
        iteration need to be the exact same. So unless you can compute
        everything, skip evaluation.

        A common example for why you might want to skip evaluation is that at
        the beginning of training, you may not have enough data for a
        validation and training set.

        :return:
        """
        return (
            len(self._exploration_paths) > 0
            and self.replay_buffer.num_steps_can_sample() >= self.min_steps_before_training
        )

    def _can_train(self):
        return self.replay_buffer.num_steps_can_sample() >= self.min_steps_before_training

    def _get_action_and_info(self, observation):
        """
        Get an action to take in the environment.
        :param observation:
        :return:
        """
        self.exploration_policy.set_num_steps_total(self._n_env_steps_total)
        return self.exploration_policy.get_action(
            observation,
        )

    def _start_epoch(self, epoch):
        self._epoch_start_time = time.time()
        self._exploration_paths = []
        self._do_train_time = 0
        logger.push_prefix('Iteration #%d | ' % epoch)

    def _end_epoch(self):
        self.eval_statistics = None
        logger.log("Epoch Duration: {0}".format(
            time.time() - self._epoch_start_time
        ))
        logger.log("Started Training: {0}".format(self._can_train()))
        logger.pop_prefix()

    def _start_new_rollout(self):
        self.exploration_policy.reset()
        return self.training_env.reset()

    def _handle_path(self, path):
        """
        Naive implementation: just loop through each transition.
        :param path:
        :return:
        """
        for (
            ob,
            action,
            reward,
            next_ob,
            terminal,
            agent_info,
            env_info
        ) in zip(
            path["observations"],
            path["actions"],
            path["rewards"],
            path["next_observations"],
            path["terminals"],
            path["agent_infos"],
            path["env_infos"],
        ):
            self._handle_step(
                ob,
                action,
                reward,
                next_ob,
                terminal,
                agent_info=agent_info,
                env_info=env_info,
            )
        self._handle_rollout_ending()

    def _handle_step(
        self,
        observation,
        action,
        reward,
        next_observation,
        terminal,
        absorbing,
        agent_info,
        env_info,
    ):
        """
        Implement anything that needs to happen after every step
        :return:
        """
        self._current_path_builder.add_all(
            observations=observation,
            actions=action,
            rewards=reward,
            next_observations=next_observation,
            terminals=terminal,
            absorbing=absorbing,
            agent_infos=agent_info,
            env_infos=env_info,
        )
        self.replay_buffer.add_sample(
            observation=observation,
            action=action,
            reward=reward,
            terminal=terminal,
            next_observation=next_observation,
            absorbing=absorbing,
            agent_info=agent_info,
            env_info=env_info,
        )

    def _handle_rollout_ending(self):
        """
        Implement anything that needs to happen after every rollout.
        """
        self.replay_buffer.terminate_episode()
        self._n_rollouts_total += 1
        if len(self._current_path_builder) > 0:
            self._exploration_paths.append(
                self._current_path_builder
            )
            self._current_path_builder = PathBuilder()

    def get_epoch_snapshot(self, epoch):
        """
        Probably will be overridden by each algorithm
        """
        data_to_save = dict(
            epoch=epoch,
            exploration_policy=self.exploration_policy,
        )
        if self.save_environment:
            data_to_save['env'] = self.training_env
        return data_to_save

    # @abc.abstractmethod
    # def load_snapshot(self, snapshot):
    #     """
    #     Should be implemented on a per algorithm basis
    #     taking into consideration the particular
    #     get_epoch_snapshot implementation for the algorithm
    #     """
    #     pass

    def get_extra_data_to_save(self, epoch):
        """
        Save things that shouldn't be saved every snapshot but rather
        overwritten every time.
        :param epoch:
        :return:
        """
        if self.render:
            self.training_env.render(close=True)
        data_to_save = dict(
            epoch=epoch,
        )
        if self.save_environment:
            data_to_save['env'] = self.training_env
        if self.save_replay_buffer:
            data_to_save['replay_buffer'] = self.replay_buffer
        if self.save_algorithm:
            data_to_save['algorithm'] = self
        return data_to_save

    @abc.abstractmethod
    def training_mode(self, mode):
        """
        Set training mode to `mode`.
        :param mode: If True, training will happen (e.g. set the dropout
        probabilities to not all ones).
        """
        pass

    @abc.abstractmethod
    def _do_training(self):
        """
        Perform some update, e.g. perform one gradient step.
        :return:
        """
        pass

    def evaluate(self, epoch):
        """
        Evaluate the policy, e.g. save/print progress.
        :param epoch:
        :return:
        """
        statistics = OrderedDict()
        try:
            statistics.update(self.eval_statistics)
            self.eval_statistics = None
        except:
            print('No Stats to Eval')

        logger.log("Collecting samples for evaluation")
        test_paths = self.eval_sampler.obtain_samples()

        statistics.update(eval_util.get_generic_path_information(
            test_paths, stat_prefix="Test",
        ))
        statistics.update(eval_util.get_generic_path_information(
            self._exploration_paths, stat_prefix="Exploration",
        ))

        if hasattr(self.env, "log_diagnostics"):
            self.env.log_diagnostics(test_paths)
        if hasattr(self.env, "log_statistics"):
            statistics.update(self.env.log_statistics(test_paths))
        if epoch % self.freq_log_visuals == 0:
            if hasattr(self.env, "log_visuals"):
                self.env.log_visuals(test_paths, epoch, logger.get_snapshot_dir())

        average_returns = eval_util.get_average_returns(test_paths)
        statistics['AverageReturn'] = average_returns
        for key, value in statistics.items():
            logger.record_tabular(key, value)

        best_statistic = statistics[self.best_key]
        if best_statistic > self.best_statistic_so_far:
            self.best_statistic_so_far = best_statistic
            if self.save_best and epoch >= self.save_best_starting_from_epoch:
                data_to_save = {
                    'epoch': epoch,
                    'statistics': statistics
                }
                data_to_save.update(self.get_epoch_snapshot(epoch))
                logger.save_extra_data(data_to_save, 'best.pkl')
                print('\n\nSAVED BEST\n\n')

max_stars_repo_name: apexrl/EBIL-torch

max_issues_repo_name: apexrl/EBIL-torch

max_forks_repo_name: apexrl/EBIL-torch


text:
"""
Metric function library
Author: Yunpeng Chen
* most of the code are inspired by MXNet
"""
import logging
import numpy as np
import torch
import torch.nn as nn
import torch.nn.functional as F

class EvalMetric(object):

	def __init__(self, name, **kwargs):
		self.name = str(name)
		self.reset()

	def update(self, preds, labels, losses):
		raise NotImplementedError()

	def reset(self):
		self.num_inst = 0
		self.sum_metric = 0.0

	def get(self):
		if self.num_inst == 0:
			return (self.name, float('nan'))
		else:
			return (self.name, self.sum_metric / self.num_inst)

	def get_name_value(self):
		name, value = self.get()
		if not isinstance(name, list):
			name = [name]
		if not isinstance(value, list):
			value = [value]
		return list(zip(name, value))

	def check_label_shapes(self, preds, labels):
		# raise if the shape is inconsistent
		if (type(labels) is list) and (type(preds) is list):
			label_shape, pred_shape = len(labels), len(preds)
		else:
			label_shape, pred_shape = labels.shape[0], preds.shape[0]

		if label_shape != pred_shape:
			raise NotImplementedError("")

class MetricList(EvalMetric):
	"""Handle multiple evaluation metric
	"""
	def __init__(self, *args, name="metric_list"):
		assert all([issubclass(type(x), EvalMetric) for x in args]), \
			"MetricList input is illegal: {}".format(args)
		self.metrics = [metric for metric in args]
		super(MetricList, self).__init__(name=name)

	def update(self, preds, labels, losses=None):
		preds = [preds] if type(preds) is not list else preds
		labels = [labels] if type(labels) is not list else labels
		losses = [losses] if type(losses) is not list else losses

		for metric in self.metrics:
			metric.update(preds, labels, losses)

	def reset(self):
		if hasattr(self, 'metrics'):
			for metric in self.metrics:
				metric.reset()
		else:
			logging.warning("No metric defined.")

	def get(self):
		ouputs = []
		for metric in self.metrics:
			ouputs.append(metric.get())
		return ouputs

	def get_name_value(self):
		ouputs = []
		for metric in self.metrics:
			ouputs.append(metric.get_name_value())
		return ouputs

####################
# COMMON METRICS
####################

class Accuracy(EvalMetric):
	"""Computes accuracy classification score.
	"""
	def __init__(self, name='accuracy', topk=1):
		super(Accuracy, self).__init__(name)
		self.topk = topk

	def update(self, preds, labels, losses):
		preds = [preds] if type(preds) is not list else preds
		labels = [labels] if type(labels) is not list else labels

		self.check_label_shapes(preds, labels)
		for pred, label in zip(preds, labels):
			assert self.topk <= pred.shape[1], \
				"topk({}) should no larger than the pred dim({})".format(self.topk, pred.shape[1])
			_, pred_topk = pred.topk(self.topk, 1, True, True)

			pred_topk = pred_topk.t()
			correct = pred_topk.eq(label.view(1, -1).expand_as(pred_topk))

			self.sum_metric += float(correct.view(-1).float().sum(0, keepdim=True).numpy())
			self.num_inst += label.shape[0]

class Loss(EvalMetric):
	"""Dummy metric for directly printing loss.
	"""
	def __init__(self, name='loss'):
		super(Loss, self).__init__(name)

	def update(self, preds, labels, losses):
		assert losses is not None, "Loss undefined."
		for loss in losses:
			self.sum_metric += float(loss.numpy().sum())
			self.num_inst += loss.numpy().size

# discrepancy loss used in MCD (CVPR 18)
def dis_MCD(out1, out2):
	return torch.mean(torch.abs(F.softmax(out1,dim=1) - F.softmax(out2, dim=1)))

# correlation norm related losses
def hard_norm_dis(corr, norm_type='nuc', dim=(1,2), norm_tgt=25, norm_dis_weight=0.05, dis_type='l2'):
	if type(dim) is tuple:
		norm = torch.norm(corr, p=norm_type, dim=dim)
	else:
		norm = torch.norm(corr, p=norm_type)
	if dis_type == 'l2':
		dis_loss = (norm.mean() - norm_tgt) ** 2 # L2 distance with norm target
	elif dis_type == 'smoothl1':
		norm_tgt = norm_tgt + torch.zeros(norm.shape[0])
		if torch.cuda.is_available():
			norm_tgt = norm_tgt.cuda()
		loss = nn.SmoothL1Loss()
		dis_loss = loss(norm, norm_tgt)
	return norm_dis_weight * dis_loss

def soft_norm_dis(corr, norm_type='nuc', dim=(1,2), norm_dis_weight=0.05):
	if type(dim) is tuple:
		norm_tgt = torch.norm(corr, p=norm_type, dim=dim).detach()
		norm = torch.norm(corr, p=norm_type, dim=dim)
	else:
		norm_tgt = torch.norm(corr, p=norm_type).detach()
		norm = torch.norm(corr, p=norm_type)
	assert norm_tgt.requires_grad == False
	norm_tgt = norm_tgt + 2.0
	dis_loss = ((norm - norm_tgt) ** 2).mean()
	return norm_dis_weight * dis_loss # L2 distance with norm target

def bnm_loss(tgt_feat, bnm_loss_weight=1.0):
	softmax_tgt = nn.Softmax(dim=1)(tgt_feat)
	loss_BNM = -torch.norm(softmax_tgt,'nuc')
	return bnm_loss_weight * loss_BNM

def guassian_kernel(source, target, kernel_mul=2.0, kernel_num=5, fix_sigma=None):
	n_samples = int(source.size()[0])+int(target.size()[0])
	total = torch.cat([source, target], dim=0)
	total0 = total.unsqueeze(0).expand(int(total.size(0)), int(total.size(0)), int(total.size(1))) # For feature MMD, until size(1)
	total1 = total.unsqueeze(1).expand(int(total.size(0)), int(total.size(0)), int(total.size(1))) # For feature MMD, until size(1)
	# total0 = total.unsqueeze(0).expand(int(total.size(0)), int(total.size(0)), int(total.size(1)), int(total.size(2))) # For corr MMD, until size(2)
	# total1 = total.unsqueeze(1).expand(int(total.size(0)), int(total.size(0)), int(total.size(1)), int(total.size(2))) # For corr MMD, until size(2)
	L2_distance = ((total0-total1)**2).sum(2) # For feature MMD, use only a single sum(2)
	# L2_distance = ((total0-total1)**2).sum(2).sum(2) # For corr MMD, use two sum(2)
	if fix_sigma:
		bandwidth = fix_sigma
	else:
		bandwidth = torch.sum(L2_distance.data) / (n_samples**2-n_samples)
	bandwidth /= kernel_mul ** (kernel_num // 2)
	bandwidth_list = [bandwidth * (kernel_mul**i) for i in range(kernel_num)]
	kernel_val = [torch.exp(-L2_distance / bandwidth_temp) for bandwidth_temp in bandwidth_list]
	return sum(kernel_val)

def CORAL(source, target):
	d = source.data.shape[1]
	# source covariance
	xm = torch.mean(source, 0, keepdim=True) - source
	xc = xm.t() @ xm
	# target covariance
	xmt = torch.mean(target, 0, keepdim=True) - target
	xct = xmt.t() @ xmt
	# frobenius norm between source and target
	loss = torch.mean(torch.mul((xc - xct), (xc - xct)))
	loss = loss/(4*d*d)
	return loss

def mmd_rbf(source, target, kernel_mul=2.0, kernel_num=5, fix_sigma=None, ver=2):
	batch_size = int(source.size()[0])
	kernels = guassian_kernel(source, target, kernel_mul=kernel_mul, kernel_num=kernel_num, fix_sigma=fix_sigma)

	loss = 0

	if ver==1:
		for i in range(batch_size):
			s1, s2 = i, (i + 1) % batch_size
			t1, t2 = s1 + batch_size, s2 + batch_size
			loss += kernels[s1, s2] + kernels[t1, t2]
			loss -= kernels[s1, t2] + kernels[s2, t1]
		loss = loss.abs_() / float(batch_size)
	elif ver==2:
		XX = kernels[:batch_size, :batch_size]
		YY = kernels[batch_size:, batch_size:]
		XY = kernels[:batch_size, batch_size:]
		YX = kernels[batch_size:, :batch_size]
		loss = torch.mean(XX + YY - XY - YX)
	else:
		raise ValueError('ver == 1 or 2')

	return loss

def DAN(source, target, kernel_mul=2.0, kernel_num=5, fix_sigma=None):
	batch_size = int(source.size()[0])
	kernels = guassian_kernel(source, target, kernel_mul=kernel_mul, kernel_num=kernel_num, fix_sigma=fix_sigma)

	loss1 = 0
	for s1 in range(batch_size):
		for s2 in range(s1+1, batch_size):
			t1, t2 = s1+batch_size, s2+batch_size
			loss1 += kernels[s1, s2] + kernels[t1, t2]
	loss1 = loss1 / float(batch_size * (batch_size - 1) / 2)

	loss2 = 0
	for s1 in range(batch_size):
		for s2 in range(batch_size):
			t1, t2 = s1+batch_size, s2+batch_size
			loss2 -= kernels[s1, t2] + kernels[s2, t1]
	loss2 = loss2 / float(batch_size * batch_size)
	return loss1 + loss2

def JAN(source_list, target_list, kernel_muls=[2.0, 2.0], kernel_nums=[2, 5], fix_sigma_list=[None, None], ver=2):
	batch_size = int(source_list[0].size()[0])
	layer_num = len(source_list)
	joint_kernels = None
	for i in range(layer_num):
		source = source_list[i]
		target = target_list[i]
		kernel_mul = kernel_muls[i]
		kernel_num = kernel_nums[i]
		fix_sigma = fix_sigma_list[i]
		kernels = guassian_kernel(source, target,
								  kernel_mul=kernel_mul, kernel_num=kernel_num, fix_sigma=fix_sigma)
		if joint_kernels is not None:
			joint_kernels = joint_kernels * kernels
		else:
			joint_kernels = kernels

	loss = 0

	if ver==1:
		for i in range(batch_size):
			s1, s2 = i, (i + 1) % batch_size
			t1, t2 = s1 + batch_size, s2 + batch_size
			loss += joint_kernels[s1, s2] + joint_kernels[t1, t2]
			loss -= joint_kernels[s1, t2] + joint_kernels[s2, t1]
		loss = loss.abs_() / float(batch_size)
	elif ver==2:
		XX = joint_kernels[:batch_size, :batch_size]
		YY = joint_kernels[batch_size:, batch_size:]
		XY = joint_kernels[:batch_size, batch_size:]
		YX = joint_kernels[batch_size:, :batch_size]
		loss = torch.mean(XX + YY - XY - YX)
	else:
		raise ValueError('ver == 1 or 2')

	return loss

if __name__ == "__main__":
	import torch

	# Test Accuracy
	predicts = [torch.from_numpy(np.array([[0.3, 0.7], [0, 1.], [0.4, 0.6]]))]
	labels   = [torch.from_numpy(np.array([   0,            1,          1 ]))]
	losses   = [torch.from_numpy(np.array([   0.3,       0.4,       0.5   ]))]

	logging.getLogger().setLevel(logging.DEBUG)
	logging.debug("input pred:  {}".format(predicts))
	logging.debug("input label: {}".format(labels))
	logging.debug("input loss: {}".format(labels))

	acc = Accuracy()

	acc.update(preds=predicts, labels=labels, losses=losses)

	logging.info(acc.get())

	# Test MetricList
	metrics = MetricList(Loss(name="ce-loss"),
						 Accuracy(topk=1, name="acc-top1"),
						 Accuracy(topk=2, name="acc-top2"),
						 )
	metrics.update(preds=predicts, labels=labels, losses=losses)

	logging.info("------------")
	logging.info(metrics.get())
	acc.get_name_value()


text:
import numpy as np
import oapackage as oa
import mldoe.design as des

# HELPER FUNCTIONS ------------------------------------------------------
def nauty_reduction(candi_set: list, resolution: int = 3) -> list:
    """
    Remove isomorphism design from a list of candidate designs.

    :param candi_set: list of candidate designs
    :type candi_set: List[design]
    :param resolution: minimal resolution for the candidate designs
    :type resolution: int
    :return: list of the non-isomorphic designs from the input set
    """
    # Turn all arrays into array_link form
    al = [oa.array_link(d.array)
          for d in candi_set if d.resolution >= resolution]
    # Define the isomorphism classes
    index, _ = selectIsomorphismClasses(al, verbose=0)
    # Select one rep. per class
    _, zz = np.unique(index, return_index=True)
    zz.sort()
    return [candi_set[idx] for idx in list(zz)]

def selectIsomorphismClasses(sols: list, verbose: int = 0) -> tuple:
    """
    Determine the unique isomorphism classes among a set of designs and attribute a class to each member of the set.
    :param sols: list of candidate designs to be sorted
    :type sols: List[design]
    :param verbose: verbosity level
    :type verbose: int
    :return: tuple of the indices and the designs in their canonical form
    :rtype: tuple(List[int],List[np.array])
    """
    # Perform check on array data type
    mm = []
    for ii, al in enumerate(sols):
        if verbose:
            oa.tprint(
                'selectIsomorphismClasses: process array %d/%d' % (ii, len(sols)), dt=4)
        # al = oa.makearraylink(al)
        tt = oa.reduceOAnauty(al)
        alx = tt.apply(al)
        mm.append(np.array(alx))

    # Perform uniqueness check
    nn = len(mm)
    qq = np.array([None] * nn, dtype=object)
    for ii in range(nn):
        qq[ii] = mm[ii].flatten()
    _, indices = np.unique(np.vstack(qq), axis=0, return_inverse=True)
    if verbose >= 1:
        print('selectIsomorphismClasses: %d reduced to %d' %
              (len(sols), np.unique(indices).size))
    return indices, mm

# --------------------------------------------------------------------

def main():
    # Create a 64-run mixed-level design with only basic factors and a single added factor, with a single four-level
    # factor constructed from the three pseudo factors 1,2,3 (constructed using the grouping scheme of Wu 1989).
    root_set = [des.MLD(128, [[1, 2, 3]], [4, 8, 16, i])
                for i in range(1, 32) if i not in [1, 2, 3, 4, 8, 16]]
    parent_set = nauty_reduction(root_set)

    # Initialize candidate list
    candidate_lst = []
    min_res = 4
    # Find candidates for each parent
    for parent in parent_set:
        # Consider generators that have length >= 3 as added factors
        gen_list = parent.generators(resolution=min_res)
        # For each generator, compute the DOP and check that the parent design has MA over all DOP
        parent_wlp = parent.wlp
        for gen in gen_list:
            candidate = des.MLD(
                parent.n_runs, parent.pf_lst, parent.cols + [gen])
            MA = True
            for dop in candidate.dop():
                dop_wlp = dop.wlp
                for i, x in enumerate(dop_wlp):
                    if parent_wlp[i] < dop_wlp[i]:
                        break
                    elif parent_wlp[i] > dop_wlp[i]:
                        MA = False
                        break
                if not MA:
                    break
            if MA:
                candidate_lst.append(candidate)

    # Run the list through the isomorphism selection algorithm
    iso_lst = nauty_reduction(candidate_lst, min_res)
    print(f'{len(parent_set)} parent designs')
    print(f'{len(candidate_lst)} candidates with resolution <= {min_res} generated')
    print(f'{len(iso_lst)} non-isomorphic designs found')

if __name__ == "__main__":
    import cProfile
    with cProfile.Profile() as pr:
        main()
    pr.print_stats(sort='cumulative')

max_stars_repo_name: ABohynDOE/mldoe

max_issues_repo_name: ABohynDOE/mldoe

max_forks_repo_name: ABohynDOE/mldoe


text:
import numpy as np

def get_max_F1(F1_score: list):
    if len(F1_score) > 0:
        max_F1 = np.max(F1_score)
        optimal_peak_number = np.min(np.where(F1_score == max_F1)[0])
    else:
        max_F1 = 0
        optimal_peak_number = np.nan
    return max_F1, optimal_peak_number

def get_clean_points_close2point(point, clean, radius):
    close_to_point = []
    distances = []
    for clean_p in clean:
        dist = np.linalg.norm(clean_p - point)
        if dist <= radius:
            close_to_point.append(clean_p)
            distances.append(dist)
    close_to_point = [tuple(p) for p in close_to_point]
    return close_to_point, distances

def precision_recall_calculator(predicted_coordinates: np.array or list,
                                value_predicted: list,
                                true_coordinates: np.array or list,
                                radius: float):
    true_coordinates = list(true_coordinates)
    predicted_coordinates = list(predicted_coordinates)
    detected_true = list()
    predicted_true_positives = list()
    predicted_redundant = list()
    value_predicted_true_positives = list()
    value_predicted_redundant = list()
    precision = list()
    recall = list()
    total_true_points = len(true_coordinates)
    assert total_true_points > 0, "one empty list here!"
    if len(predicted_coordinates) == 0:
        print("No predicted points")
        precision = []
        recall = []
        detected_true = []
        predicted_true_positives = []
        predicted_false_positives = []
        value_predicted_true_positives = []
        value_predicted_false_positives = []
        predicted_redundant = []
        value_predicted_redundant = []
        false_negatives = total_true_points
    else:
        predicted_false_positives = list()
        value_predicted_false_positives = list()
        for value, point in zip(value_predicted, predicted_coordinates):
            close_to_point, distances = get_clean_points_close2point(
                point,
                true_coordinates,
                radius
            )
            if len(close_to_point) > 0:
                flag = "true_positive_candidate"
                flag_tmp = "not_redundant_yet"
                for dist, clean_p in sorted(zip(distances, close_to_point)):
                    if flag == "true_positive_candidate":
                        if tuple(clean_p) not in detected_true:
                            detected_true.append(tuple(clean_p))
                            flag = "true_positive"
                        else:
                            flag_tmp = "redundant_candidate"
                    # else:
                    # print(point, "is already flagged as true positive")
                if flag == "true_positive":
                    predicted_true_positives.append(tuple(point))
                    value_predicted_true_positives.append(value)
                elif flag == "true_positive_candidate" and \
                        flag_tmp == "redundant_candidate":
                    predicted_redundant.append(tuple(point))
                    value_predicted_redundant.append(value)
                else:
                    print("This should never happen!")
            else:
                predicted_false_positives.append(tuple(point))
                value_predicted_false_positives.append(value)
            true_positives_total = len(predicted_true_positives)
            false_positives_total = len(predicted_false_positives)
            total_current_predicted_points = true_positives_total + \
                                             false_positives_total
            precision.append(true_positives_total / total_current_predicted_points)
            recall.append(true_positives_total)
        false_negatives = [point for point in true_coordinates if tuple(point) not in detected_true]
        N_inv = 1 / total_true_points
        recall = np.array(recall) * N_inv
        recall = list(recall)
    return precision, recall, detected_true, predicted_true_positives, \
           predicted_false_positives, value_predicted_true_positives, \
           value_predicted_false_positives, false_negatives, predicted_redundant, \
           value_predicted_redundant

def f1_score_calculator(precision: list, recall: list):
    f1_score = []
    if len(precision) == 0:
        print("No precision and recall")
        f1_score = [0]
    else:
        for p, r in zip(precision, recall):
            if p + r != 0:
                f1_score.append(2 * p * r / float(p + r))
            else:
                f1_score.append(0)
    return f1_score

def quadrature_calculator(x_points: list, y_points: list) -> float:
    """
    This function computes an approximate value of the integral of a real
    function f in an interval, using the trapezoidal rule.

    Input:
    x_points: is a list of points in the x axis (not necessarily ordered)
    y_points: is a list of points, such that y_points[n] = f(x_points[n]) for
    each n.
    """
    # sorted_y = [p for _, p in sorted(zip(x_points, y_points))]
    sorted_y = [p for _, p in
                sorted(list(zip(x_points, y_points)), key=lambda x: x[0])]
    n = len(y_points)
    sorted_x = sorted(x_points)

    trapezoidal_rule = [
        0.5 * (sorted_x[n + 1] - sorted_x[n]) * (sorted_y[n + 1] + sorted_y[n])
        for n in range(n - 1)]

    return float(np.sum(trapezoidal_rule))

def pr_auc_score(precision: list, recall: list) -> float:
    """
    This function computes an approximate value to the area
    under the precision-recall (PR) curve.
    """
    return quadrature_calculator(recall, precision)

max_stars_repo_name: mrmattuschka/DeePiCt

max_issues_repo_name: mrmattuschka/DeePiCt

max_forks_repo_name: ZauggGroup/DeePiCt


text:
import json
import zlib
import time

import numpy as np

from rp_server import TcpServer
from rp_server import DetectronWrapper
from rp_server import bin2img, bin2int, int2bin

def magic(idx):
    foo = lambda x: idx + 1 if x else 0
    return lambda x: map(foo, x)

def pack_instseg_(resp):
    """
    Pack detectron2 (Inst_seg) result data into binary package
    """
    boxes = resp.pred_boxes.tensor.cpu().detach().numpy().tolist()
    scores = resp.scores.cpu().detach().numpy().tolist()
    classes = resp.pred_classes.cpu().detach().numpy().tolist()
    masks = resp.pred_masks.cpu().detach().numpy()

    # sent a 0 size package
    if boxes is None:
        return int2bin(0)

    # fused all masks into a single mask
    # mask_id = idx + 1
    start_t = time.time()
    n_inst, h, w = masks.shape
    fused_masks = np.zeros(shape=(h, w), dtype="uint8")

    # start_t = time.time()
    for idx in range(n_inst):
        for i in range(h):
            for j in range(w):
                if masks[idx, i, j]:
                    fused_masks[i, j] = idx + 1

    # for idx in range(n_inst):
    #     map(magic(idx), masks[idx])
    #     # a = [[*m] for m in [*map(magic(idx), masks[idx])]]
    #     # fused_masks += np.array([list(m) for m in list(map(magic(idx), masks[idx]))], dtype="uint8")
    # print("Time elapsed: {}".format(time.time() - start_t))

    compressed_mask_bin = zlib.compress(fused_masks.tobytes())
    mask_size_bin = int2bin( len(compressed_mask_bin) )
    mask_pkg = mask_size_bin + int2bin(w) + int2bin(h) + compressed_mask_bin

    info_json = {
        "boxes": boxes,
        "scores": scores,
        "classes": classes
    }
    info_bin = json.dumps(info_json).encode()

    pkg_size = int2bin(len(mask_pkg) + len(info_bin))

    return pkg_size + mask_pkg + info_bin

def pack_panoseg_(resp):
    """
    Pack detectron2 (Pano_seg) result data into binary

    | pkg_size (4B int) | map_size (4B int) | width (4B int) | ...
    | height (4B int) | binary_map (map_size B) | json_info_binary (rest) |
    """
    # pack segmentation map
    pano_resp = resp["panoptic_seg"]
    inst_resp = resp["instances"]
    seg_map = pano_resp[0].cpu().detach().numpy().astype("uint8")
    h, w = seg_map.shape
    compressed_map_bin = zlib.compress(seg_map.tobytes())
    map_size_bin = int2bin( len(compressed_map_bin) )
    map_data_pkg = map_size_bin + int2bin(w) + int2bin(h) + compressed_map_bin

    # pack semantic information
    info_json = {
        "info": pano_resp[1],
        "boxes": inst_resp.pred_boxes.tensor.cpu().detach().numpy().tolist()
    }
    info_bin = json.dumps(info_json).encode()

    # total package size
    print(len(map_data_pkg))
    pkg_size = int2bin(len(map_data_pkg) + len(info_bin))

    return pkg_size + map_data_pkg + info_bin

DT_ENCODER = {
    "Pano_seg": pack_panoseg_,
    "Inst_seg": pack_instseg_
}

class DetectronServer(TcpServer):

    def __init__(self, host, port, model_type="Pano_seg"):
        super(DetectronServer, self).__init__(host=host, port=port)

        self.dt_ = DetectronWrapper(task=model_type)
        self.model_ = model_type

    def handle_connection_(self, conn, addr):
        conn_id = "{}:{}".format(addr[0], addr[1])
        print('New connection from {}'.format(conn_id))

        while not self.quit_event_.is_set():
            pack_size = conn.recv(4)

            # end of Connection
            if not pack_size:
                break

            pack_size = bin2int(pack_size)
            # fetch data package
            data = self.recv_all_(conn, pack_size)

            img = bin2img(data)
            ret = self.dt_.predict(img)

            # send back response
            conn.sendall( self.pack_(ret, self.model_) )

        conn.close()
        print("Connection {}: closed".format(conn_id))

    def pack_(self, resp, model_type):
        if model_type in DT_ENCODER:
            return DT_ENCODER[model_type](resp)
        else:
            raise Exception("Does not support type: {}".format(model_type))

if __name__ == "__main__":
    server = DetectronServer(host="192.168.1.94", port=8801, model_type="Pano_seg")
    server.launch()

# pkg = {
#     "boxes": boxes,
#     "scores": scores,
#     "classes": classes,
#     # "masks": masks
# }


text:
#from scipy.io import arff #String attributes not supported by scipy.io arff...
# from Lib import *

import arff
import pandas as pd

class FileLoader(object):
    '''
    classdocs
    '''

    def __init__(self):
        '''
        Constructor
        '''
        file_path = 'C:/Users/Christian/Documents/GitHub/ML_FaultUnderstanding/data/'
        self.file_1 = file_path  + 'consolidated_Final_Experiment_1.arff'
        self.file_2 = file_path  + 'consolidated_Final_Experiment_2.arff'

    def _load_file_1(self):
        return self.__load__file(self.file_1)

    def _load_file_2(self):
        return self.__load__file(self.file_2)

    'Returns a panda Dataframe'
    def __load__file(self,file_path):
        file_object = open(file_path)
        file_content = file_object.read()
        dataset = arff.loads(file_content,encode_nominal=True,return_type=arff.DENSE)
        #print(dataset['description'])
        #print(dataset['relation'])
        #print(dataset['attributes'])
        #print(dataset['data'][0])
        #print("number of imported lines: " + str(dataset['data'].__len__()))

        #label data frame columns
        #https://pandas.pydata.org/pandas-docs/stable/10min.html
        df = pd.DataFrame(dataset['data'])
        df_labels = pd.DataFrame(dataset['attributes'])
        df.columns = df_labels[0]
        return df

reason: from scipy


text:
"""Tests for util_plots."""

from common_python.plots import util_plots
import common_python.constants as cn

import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
import unittest

IGNORE_TEST = False
IS_PLOT = False

DF = pd.DataFrame({
  'a': [-1, 0, 1],
  'b': [0, 1, -1],
  'c': [1, -1, 0],
  })

class TestFunction(unittest.TestCase):

  def testPlotTrinaryHeatmap(self):
    # Smoke tests
    if IGNORE_TEST:
      return
    util_plots.plotTrinaryHeatmap(DF, is_plot=IS_PLOT)
    plt.figure()
    ax = plt.gca()
    util_plots.plotTrinaryHeatmap(DF, ax=ax, is_plot=IS_PLOT)

  def testPlotCategoricalHeatmap(self):
    # Smoke tests
    if IGNORE_TEST:
      return
    df = pd.DataFrame({
        'a': [0, 0, 1],
        'b': [0, 1, 0],
        'c': [0, 1, 0],
        })
    df.index = ['x', 'y', 'z']
    opts = {
        cn.PLT_TITLE: "test",
        cn.PLT_XLABEL: "x",
        cn.PLT_YLABEL: "y",
        }
    util_plots.plotCategoricalHeatmap(df, is_plot=IS_PLOT, **opts)

  def testPlotCorr(self):
    # Smoke tests
    if IGNORE_TEST:
      return
    size = 10
    df = pd.DataFrame(np.random.rand(size, size))
    util_plots.plotCorr(df, is_plot=IS_PLOT, xlabel="x", title="title")

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

max_stars_repo_name: ModelEngineering/topics-course

max_issues_repo_name: ModelEngineering/topics-course

max_forks_repo_name: ModelEngineering/topics-course