test_data_utils.py

# imports

# standard
    
import random

import math

from statistics import mean

# local, internal

from src.topupopt.data.misc import utils

#******************************************************************************
#******************************************************************************

class TestDataUtils:
    
    # *************************************************************************
    # *************************************************************************

    def test_profile_synching2(self):
        
        integration_result = 10446
        ratio_min_avg = 0.2
        min_to_max_ratio = ratio_min_avg/(2-ratio_min_avg)
        
        avg_state = [
            2.66,
            2.34,
            3.54,
            7.42,
            11.72,
            16.94,
            17.94,
            17.98,
            14.1,
            10.48,
            6.74,
            3.16]
        
        time_interval_durations = [
            31, # jan
            28, # fev
            31, # mar
            30, # apr
            31, # may
            30, # june
            31, # july
            31, # august
            30, # september
            31, # october
            30, # november
            31  # december
            ] 
        
        # *********************************************************************
        # *********************************************************************
        
        # state correlates with output
        
        new_profile = utils.create_profile_using_time_weighted_state(
            integration_result=integration_result, 
            avg_state=avg_state, 
            time_interval_durations=time_interval_durations, 
            min_to_max_ratio=min_to_max_ratio,
            state_correlates_with_output=False
            )
        
        expected_result = [
            1500.0513102636057, 
            1436.2189684321309, 
            1500.0513102636044, 
            1206.6051909345115, 
            896.2493366213225, 
            525.7218723351705, 
            283.3475134825171, 
            185.83058429361876, 
            270.8127439165165, 
            538.2566419011698, 
            861.4985340132666, 
            1241.355993542566
            ]
        
        abs_tol = 1e-3
        
        assert math.isclose(
            sum(new_profile),
            integration_result,
            abs_tol=abs_tol
            )   
        
        for sample, expected_sample in zip(new_profile, expected_result):
            
            assert math.isclose(
                sample,
                expected_sample,
                abs_tol=abs_tol
                )
            
        # *********************************************************************
        # *********************************************************************
        
        # state does not correlate with output
        
        # state correlates with output
        
        new_profile = utils.create_profile_using_time_weighted_state(
            integration_result=integration_result, 
            avg_state=avg_state, 
            time_interval_durations=time_interval_durations, 
            min_to_max_ratio=min_to_max_ratio,
            state_correlates_with_output=True
            )
        
        expected_result = [
            274.3377308322865, 
            166.45500417060902, 
            274.33773083228533, 
            510.54549399699533, 
            878.1397044745678, 
            1191.4288125963367, 
            1491.041527613374, 
            1588.5584568022714, 
            1446.3379410149894, 
            1236.132399194721, 
            855.6521509182398, 
            533.0330475533233
            ]
        
        abs_tol = 1e-3
        
        assert math.isclose(
            sum(new_profile),
            integration_result,
            abs_tol=abs_tol
            )   
        
        for sample, expected_sample in zip(new_profile, expected_result):
            
            assert math.isclose(
                sample,
                expected_sample,
                abs_tol=abs_tol
                )
            
        # *********************************************************************
        # *********************************************************************
        
        # find out the peaks of the sinusoidal profile
            
        pmax, pmin = utils.max_min_sinusoidal_profile(
            integration_result=integration_result,
            period=sum(time_interval_durations),
            time_interval_duration=mean(time_interval_durations),
            min_to_max_ratio=min_to_max_ratio
            )
        
        expected_pmax, expected_pmin = 1558.972133279683, 182.02786672031687
            
        assert math.isclose(
            pmax,
            expected_pmax,
            abs_tol=1e-3
            )
            
        assert math.isclose(
            pmin,
            expected_pmin,
            abs_tol=1e-3
            )
            
        # *********************************************************************
        # *********************************************************************
        
        # raise exception
        
        error_triggered = False
        time_interval_durations.pop(0)
        try:
            new_profile = utils.create_profile_using_time_weighted_state(
                integration_result=integration_result, 
                avg_state=avg_state, 
                time_interval_durations=time_interval_durations, 
                min_to_max_ratio=min_to_max_ratio,
                state_correlates_with_output=True
                )
        except ValueError:
            error_triggered = True
        assert error_triggered
        
        # *********************************************************************
        # *********************************************************************
        
    # *************************************************************************
    # *************************************************************************
    
    def test_profile_synching(self):
        
        # synch, normal, ex1
        
        profile = [1,2,3,4]
        reference_profile = [2,3,4,1]
        synched_profile = utils.synch_profile(profile, reference_profile)
        true_synched_profile = [2,3,4,1]
        assert repr(synched_profile) == repr(true_synched_profile)
        
        # synch, normal, ex2
        
        profile = [-2,-1,1,2,0]
        reference_profile = [2,3,4,1,5]
        synched_profile = utils.synch_profile(profile, reference_profile)
        true_synched_profile = [-1,0,1,-2,2]
        assert repr(synched_profile) == repr(true_synched_profile)
        
        # synch, alternative, ex1
        
        profile = [1,2,3,4]
        reference_profile = [2,3,4,1]
        synched_profile = utils.synch_profile(profile, reference_profile, synch=False)
        true_synched_profile = [3,2,1,4]
        assert repr(synched_profile) == repr(true_synched_profile)
        
        
    # *************************************************************************
    # *************************************************************************
    
    def test_profile_generation(self):
    
        # *********************************************************************
        
        # fixed time interval durations
        
        number_tests = 10
        
        for test_index in range(number_tests):
        
            integration_period = 365*24*3600
            
            number_intervals = random.randint(1,8760)
            
            phase_shift_radians = 2*math.pi*random.random()
            
            time_interval_durations = [
                round(integration_period/number_intervals)
                for i in range(number_intervals)
                ]
            
            integration_result = 100
            
            min_to_max_ratio = 0.2
            
            profile = utils.discrete_sinusoid_matching_integral(
                integration_result, 
                time_interval_durations, 
                min_to_max_ratio, 
                phase_shift_radians=phase_shift_radians)
            
            assert math.isclose(
                sum(profile),
                integration_result,
                abs_tol=0.01
                )
        
        # *********************************************************************
        
        # import matplotlib.pyplot as plt
        
        # # Data for plotting
        # x = [i for i in range(number_intervals)]
        # y = profile
    
        # fig, ax = plt.subplots()
        # ax.plot(x, y)
    
        # ax.set(xlabel='time (s)', ylabel='voltage (mV)',
        #         title='About as simple as it gets, folks')
        # ax.grid()
    
        # #fig.savefig("test.png")
        # plt.show()
            
        # *********************************************************************
    
        # variable time step durations
        
        number_tests = 10
        
        for test_index in range(number_tests):
        
            number_intervals = random.randint(10,8760)
        
            time_interval_durations = [
                random.random()*3.6e3
                for i in range(number_intervals)
                ]
        
            integration_period = sum(time_interval_durations)
            
            phase_shift_radians = 2*math.pi*random.random()
            
            integration_result = 100
            
            min_to_max_ratio = 0.2
            
            profile = utils.discrete_sinusoid_matching_integral(
                integration_result, 
                time_interval_durations, 
                min_to_max_ratio, 
                phase_shift_radians=phase_shift_radians)
            
            assert math.isclose(
                sum(profile),
                integration_result,
                abs_tol=0.01
                )
        
        # *********************************************************************
        
        # # import matplotlib.pyplot as plt
        
        # t = [sum(time_interval_durations[0:i])
        #      for i in range(len(time_interval_durations)+1)]
        
        # # Data for plotting
        # x = [(t[i+1]+t[i])*0.5
        #      for i in range(number_intervals)] # time interval's center point
        # y = profile
    
        # fig, ax = plt.subplots()
        # ax.plot(x, y)
    
        # ax.set(xlabel='time (s)', ylabel='voltage (mV)',
        #         title='About as simple as it gets, folks')
        # ax.grid()
    
        # #fig.savefig("test.png")
        # plt.show()
    
        # *********************************************************************
        
        # use the default phase shift
        
        integration_period = 365*24*3600
        
        number_intervals = random.randint(1,8760)
        
        time_interval_durations = [
            round(integration_period/number_intervals)
            for i in range(number_intervals)
            ]
        
        integration_result = 100
        
        min_to_max_ratio = 0.2
        
        profile = utils.discrete_sinusoid_matching_integral(
            integration_result, 
            time_interval_durations, 
            min_to_max_ratio)
        
        assert math.isclose(
            sum(profile),
            integration_result,
            abs_tol=0.01
            )
        
    # *************************************************************************
    # *************************************************************************
    
    def test_key_generation(self):
        
        # generate_pseudo_unique_key
        
        key_list = (str(random.random()) for i in range(10))
        
        new_key = utils.generate_pseudo_unique_key(key_list=key_list)
        
        assert new_key not in key_list
        
        # use an empty key list
        
        new_key = utils.generate_pseudo_unique_key(key_list=[])
        
        assert new_key not in key_list
        
        # use zero iterations to force an error
        
        error_triggered = False
        try:
            new_key = utils.generate_pseudo_unique_key(key_list=key_list,
                                                       max_iterations=0)
        except Exception:
            error_triggered = True
        assert error_triggered
        
        # use a seed number to trigger more iterations
        
        import uuid
        rand = random.Random()
        rand.seed(360)
        uuid.uuid4 = lambda: uuid.UUID(int=rand.getrandbits(128), version=4)
        
        key_list = ['3e225573-4e78-48c8-bb08-efbeeb795c22',
                    'f6d30428-15d1-41e9-a952-0742eaaa5a31',
                    '8c29b906-2518-41c5-ada8-07b83508b5b8',
                    'f9a72a39-1422-4a02-af97-906ce79c32a3',
                    'b6941a48-10cc-465d-bf53-178bd2939bd1']
        
        new_key = utils.generate_pseudo_unique_key(key_list=key_list)
        
        assert new_key not in key_list
    
    #**************************************************************************
    #**************************************************************************

#******************************************************************************
#******************************************************************************