WassimTenachi / PhySO / #13

Committed 10 Jun 2024 12:28AM UTC coverage: 52.052% (-30.3%) from 82.385%

Build # #13

Build Type

push

coveralls-python

Committed by

WassimTenachi

Commit Message

Update requirements.txt

Run Details

2980 of 5725 relevant lines covered (52.05%)

0.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.2

/physo/physym/tests/dimensional_analysis_UnitTest.py

import time
import unittest
import numpy as np

from physo.physym import program as Prog
from physo.physym import library as Lib
from physo.physym import dimensional_analysis as phy
from physo.physym import vect_programs as VProg

def hard_test_case():
    # LIBRARY CONFIG
    args_make_tokens = {
                    # operations
                    "op_names"             : "all",  # or ["mul", "neg", "inv", "sin"]
                    "use_protected_ops"    : True,
                    # input variables
                    "input_var_ids"        : {"x" : 0         , "v" : 1          , "t" : 2,        },
                    "input_var_units"      : {"x" : [1, 0, 0] , "v" : [1, -1, 0] , "t" : [0, 1, 0] },
                    "input_var_complexity" : {"x" : 0.        , "v" : 1.         , "t" : 0.,       },
                    # constants
                    "constants"            : {"pi" : np.pi     , "M" : 1e6       , "const1" : 1         },
                    "constants_units"      : {"pi" : [0, 0, 0] , "M" : [0, 0, 1] , "const1" : [0, 0, 0] },
                    "constants_complexity" : {"pi" : 0.        , "M" : 1.        , "const1" : 1.        },
                    # free constants
                    "free_constants"            : {"c"               },
                    "free_constants_init_val"   : {"c"  : 10.        },
                    "free_constants_units"      : {"c"  : [1, -1, 0] },
                    "free_constants_complexity" : {"c"  : 0.         },
                           }

    my_lib = Lib.Library(args_make_tokens = args_make_tokens,
                         superparent_units = [2, -2, 1], superparent_name = "y")

    # TEST PROGRAM
    test_program_str = ["mul", "mul", "M", "n2", "c", "sub", "inv", "sqrt", "sub", "const1", "div", "n2", "v", "n2",
                        "c", "cos", "div", "sub", "div", "const1", "const1", "div", "v", "c", "div", "v", "c"]
    test_program_idx = np.array([my_lib.lib_name_to_idx[tok_str] for tok_str in test_program_str])

    # ------------------- EXPECTED LIVE BEHAVIOR -------------------
    expected_constraints = np.array([
        (True, [2., -2., 1., 0., 0., 0., 0.],),
        (False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
        (False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
        (False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
        (False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
        (True, [0., 0., 0., 0., 0., 0., 0.],),
        (True, [0., 0., 0., 0., 0., 0., 0.],),
        (True, [-0., -0., -0., -0., -0., -0., -0.],),
        (True, [-0., -0., -0., -0., -0., -0., -0.],),
        (True, [-0., -0., -0., -0., -0., -0., -0.],),
        (True, [0., 0., 0., 0., 0., 0., 0.],),
        (False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
        (False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
        (True, [2., -2., 0., 0., 0., 0., 0.],),
        (True, [1., -1., 0., 0., 0., 0., 0.],),
        (True, [0., 0., 0., 0., 0., 0., 0.],),
        (True, [0., 0., 0., 0., 0., 0., 0.],),
        (False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
        (False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
        (False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
        (False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
        (True, [0., 0., 0., 0., 0., 0., 0.],),
        (False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
        (True, [1., -1., 0., 0., 0., 0., 0.],),
        (True, [0., 0., 0., 0., 0., 0., 0.],),
        (False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
        (True, [1., -1., 0., 0., 0., 0., 0.],),
    ], dtype=object)
    expected_tokens_cases_record = np.array([3, 70, 21, 72, 21, 73, 4, 5, 5, 4, 1, 70, 21, 73, 5, 1, 6, 70, 21, 21, 72,
                                             20, 70, 73, 73, 70, 73])
    expected_phy_units = np.array([x for x in expected_constraints[:, 1]], dtype=float)
    expected_is_constraining = expected_constraints[:, 0].astype(bool)

    # ------------------- EXPECTED UNITS AT FINAL STEP -------------------
    expected_constraints_final = np.array([
           (True, [ 2., -2.,  1.,  0.,  0.,  0.,  0.],),
           (True, [ 2., -2.,  1.,  0.,  0.,  0.,  0.],),
           (True, [ 0.,  0.,  1.,  0.,  0.,  0.,  0.],),
           (True, [ 2., -2.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 1., -1.,  0.,  0.,  0.,  0.,  0.],),
           (True, [-0., -0., -0., -0., -0., -0., -0.],),
           (True, [-0., -0., -0., -0., -0., -0., -0.],),
           (True, [ 0.,  0.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 0.,  0.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 0.,  0.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 0.,  0.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 2., -2.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 1., -1.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 2., -2.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 1., -1.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 0.,  0.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 0.,  0.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 0.,  0.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 0.,  0.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 0.,  0.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 0.,  0.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 0.,  0.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 1., -1.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 1., -1.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 0.,  0.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 1., -1.,  0.,  0.,  0.,  0.,  0.],),
           (True, [ 1., -1.,  0.,  0.,  0.,  0.,  0.],),
    ], dtype=object)

    expected_phy_units_final       = np.array([x for x in expected_constraints_final[:, 1]], dtype=float)
    expected_is_constraining_final = expected_constraints_final[:, 0].astype(bool)

    return (test_program_idx, my_lib, expected_tokens_cases_record, expected_phy_units, expected_is_constraining, \
           expected_phy_units_final, expected_is_constraining_final)

class DimensionalAnalysisTest(unittest.TestCase):

    def test_assign_units_bottom_up(self):
        test_program_idx, my_lib, expected_tokens_cases_record, expected_phy_units, expected_is_constraining, \
            expected_phy_units_final, expected_is_constraining_final = hard_test_case()

        test_program_length = len(test_program_idx)
        batch_size = 3
        test_programs_idx = np.tile(test_program_idx, reps=(batch_size, 1))

        my_programs = VProg.VectPrograms(batch_size=batch_size, max_time_step=test_program_length, library=my_lib, n_realizations=1)
        my_programs.set_programs(test_programs_idx)

        # ------------------- TEST BOTTOM-UP -------------------
        try:
            phy.assign_units_bottom_up(my_programs,
                                   coords_start = my_programs.coords_of_step(0),
                                   coords_end   = my_programs.coords_of_step(my_programs.max_time_step-1),)
        except:
            self.fail("Bottom up run failed.")

        # ------------------- TEST BOTTOM-UP RESULTS-------------------
        # After bottom up
        observed_phy_units_final_after_BU       = my_programs.tokens.phy_units                 [0,:].copy()
        observed_is_constraining_final_after_BU = my_programs.tokens.is_constraining_phy_units [0,:].copy()

        # ASSERT BOTTOM UP GIVES RIGHT RESULT
        bool_works = np.array_equal(expected_is_constraining_final , observed_is_constraining_final_after_BU)
        self.assertEqual(bool_works, True)
        bool_works = np.array_equal(expected_phy_units_final       , observed_phy_units_final_after_BU, equal_nan = True )
        self.assertEqual(bool_works, True)

    def test_get_required_units(self):

        # ------------------- TEST CASE -------------------
        test_program_idx, my_lib, expected_tokens_cases_record, expected_phy_units, expected_is_constraining, \
            expected_phy_units_final, expected_is_constraining_final = hard_test_case()

        test_program_length = len(test_program_idx)
        batch_size = 3
        test_programs_idx = np.tile(test_program_idx, reps=(batch_size, 1)).transpose()

        # ------------------- TRY AT STEP: assign_required_units_at_step -------------------
        try:
            batch_size_large = int(1e3)
            test_programs_idx_large = np.tile(test_program_idx, reps=(batch_size_large, 1)).transpose()
            my_programs = VProg.VectPrograms(batch_size=batch_size_large, max_time_step=test_program_length, library=my_lib, n_realizations=1)
            t0 = time.perf_counter()
            for i, idx in enumerate(test_programs_idx_large):
                # Computing units requirements
                cases_record = phy.assign_required_units_at_step(my_programs)
                # Appending new token
                # In real world run, next token should be chosen based on constraints found above
                my_programs.append(idx)
            t1 = time.perf_counter()
            marginal_time = ((t1-t0)*1e3)/(test_program_length * batch_size_large)
            print("\nRequired units time (in ideal non-mixed cases) : %f ms / step / (prog in batch) " % marginal_time)
        except:
            self.fail("Unable to run assign_required_units_at_step")

        # ------------------- TEST AT STEP: assign_required_units_at_step -------------------
        my_programs = VProg.VectPrograms(batch_size=batch_size, max_time_step=test_program_length, library=my_lib, n_realizations=1)

        observed_phy_units           = []
        observed_is_constraining     = []
        observed_tokens_cases_record = []
        for i, idx in enumerate(test_programs_idx) :
            # Computing units requirements
            cases_record = phy.assign_required_units_at_step(my_programs)
            # Saving observed results
            observed_tokens_cases_record.   append (cases_record                                 [0]  .copy())
            observed_phy_units.             append (my_programs.tokens.phy_units                 [0,i].copy())
            observed_is_constraining.       append (my_programs.tokens.is_constraining_phy_units [0,i].copy())
            # Assertions
            bool_works = np.array_equal(observed_tokens_cases_record[i], expected_tokens_cases_record[i])
            self.assertEqual(bool_works, True)
            bool_works = np.array_equal(observed_is_constraining    [i], expected_is_constraining    [i])
            self.assertEqual(bool_works, True)
            bool_works = np.array_equal(observed_phy_units          [i], expected_phy_units          [i], equal_nan = True )
            self.assertEqual(bool_works, True)
            # Appending new token
            # In real world run, next token should be chosen based on constraints found above
            my_programs.append(idx)

        # To numpy array
        observed_phy_units           = np.array(observed_phy_units)
        observed_is_constraining     = np.array(observed_is_constraining)
        observed_tokens_cases_record = np.array(observed_tokens_cases_record)

        # Final result after all idx were added (Results using assign_required_units_at_step function)
        observed_phy_units_final_w_step       = my_programs.tokens.phy_units                 [0,:].copy()
        observed_is_constraining_final_w_step = my_programs.tokens.is_constraining_phy_units [0,:].copy()

        # (Results using assign_required_units_at_step function)
        observed_phy_units_w_step           = observed_phy_units           .copy()
        observed_is_constraining_w_step     = observed_is_constraining     .copy()
        observed_tokens_cases_record_w_step = observed_tokens_cases_record .copy()

        # ------------------- TEST AT STEP: assign_required_units_at_step from scratch -------------------
        my_programs = VProg.VectPrograms(batch_size=batch_size, max_time_step=test_program_length, library=my_lib, n_realizations=1)
        my_programs.set_programs(test_programs_idx.transpose())

        phy.assign_required_units_at_step(my_programs, from_scratch=True)

       # Final result after all idx were added (Results using assign_required_units_at_step function)
        observed_phy_units_final_w_step_from_scratch       = my_programs.tokens.phy_units                 [0,:].copy()
        observed_is_constraining_final_w_step_from_scratch = my_programs.tokens.is_constraining_phy_units [0,:].copy()

        # Assert re-computing everything from_scratch gives same result
        bool_works = np.array_equal(observed_phy_units_final_w_step_from_scratch       , observed_phy_units_final_w_step       , equal_nan=True)
        self.assertEqual(bool_works, True)
        bool_works = np.array_equal(observed_is_constraining_final_w_step_from_scratch , observed_is_constraining_final_w_step)
        self.assertEqual(bool_works, True)

        # ------------------- TEST AT COORDS: assign_required_units -------------------
        my_programs = VProg.VectPrograms(batch_size=batch_size, max_time_step=test_program_length, library=my_lib, n_realizations=1)

        observed_phy_units           = []
        observed_is_constraining     = []
        observed_tokens_cases_record = []
        for i, idx in enumerate(test_programs_idx) :
            # Computing units requirements
            coords = my_programs.coords_of_step(i)[:, [0,2]] # doing 0-th and 2-th of batch but not 1-th
            cases_record = phy.assign_required_units(my_programs, coords=coords)
            # Saving observed results
            observed_tokens_cases_record.   append (cases_record                                 [0]  .copy())
            observed_phy_units.             append (my_programs.tokens.phy_units                 [0,i].copy())
            observed_is_constraining.       append (my_programs.tokens.is_constraining_phy_units [0,i].copy())
            # Assertions
            bool_works = np.array_equal(observed_tokens_cases_record[i], expected_tokens_cases_record[i])
            self.assertEqual(bool_works, True)
            bool_works = np.array_equal(observed_is_constraining    [i], expected_is_constraining    [i])
            self.assertEqual(bool_works, True)
            bool_works = np.array_equal(observed_phy_units          [i], expected_phy_units          [i], equal_nan = True )
            self.assertEqual(bool_works, True)
            # Appending new token
            # In real world run, next token should be chosen based on constraints found above
            my_programs.append(idx)

        # To numpy array
        observed_phy_units           = np.array(observed_phy_units)
        observed_is_constraining     = np.array(observed_is_constraining)
        observed_tokens_cases_record = np.array(observed_tokens_cases_record)

        # Final result after all idx were added
        observed_phy_units_final       = my_programs.tokens.phy_units                 [0,:].copy()
        observed_is_constraining_final = my_programs.tokens.is_constraining_phy_units [0,:].copy()

        # Assert that assign_required_units at coords gives same results as assign_required_units_at_step
        # Final result
        bool_works = np.array_equal(observed_phy_units_final       , observed_phy_units_final_w_step       , equal_nan=True)
        self.assertEqual(bool_works, True)
        bool_works = np.array_equal(observed_is_constraining_final , observed_is_constraining_final_w_step)
        self.assertEqual(bool_works, True)
        # Intermediate results
        bool_works = np.array_equal(observed_phy_units             , observed_phy_units_w_step             , equal_nan=True)
        self.assertEqual(bool_works, True)
        bool_works = np.array_equal(observed_is_constraining       , observed_is_constraining_w_step)
        self.assertEqual(bool_works, True)
        bool_works = np.array_equal(observed_tokens_cases_record   , observed_tokens_cases_record_w_step)
        self.assertEqual(bool_works, True)

        # ------------------- CHECK ALL THE UNITS FOUND ARE COHERENT WITH BOTTOM-UP -------------------
        try:
            phy.assign_units_bottom_up(my_programs,
                                   coords_start = my_programs.coords_of_step(0),
                                   coords_end   = my_programs.coords_of_step(my_programs.max_time_step-1),)
        except:
            self.fail("Final check bottom up failed.")

        # After bottom up
        observed_phy_units_final_after_BU       = my_programs.tokens.phy_units                 [0,:].copy()
        observed_is_constraining_final_after_BU = my_programs.tokens.is_constraining_phy_units [0,:].copy()

        # ASSERT BOTTOM UP GIVES RIGHT RESULT
        bool_works = np.array_equal(expected_is_constraining_final , observed_is_constraining_final_after_BU)
        self.assertEqual(bool_works, True)
        bool_works = np.array_equal(expected_phy_units_final       , observed_phy_units_final_after_BU, equal_nan = True )
        self.assertEqual(bool_works, True)

        # ASSERT THAT END STEP OF LIVE UNITS CONSTRAINTS GIVES SAME AS BOTTOM UP
        bool_works = np.array_equal(observed_is_constraining_final , observed_is_constraining_final_after_BU)
        self.assertEqual(bool_works, True)
        bool_works = np.array_equal(observed_phy_units_final       , observed_phy_units_final_after_BU, equal_nan = True )
        self.assertEqual(bool_works, True)

        return None


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

1	import time	1✔
2	import unittest	1✔
3	import numpy as np	1✔
4
5	from physo.physym import program as Prog	1✔
6	from physo.physym import library as Lib	1✔
7	from physo.physym import dimensional_analysis as phy	1✔
8	from physo.physym import vect_programs as VProg	×
9
10	def hard_test_case():	1✔
11	# LIBRARY CONFIG
12	args_make_tokens = {	1✔
13	# operations
14	"op_names" : "all", # or ["mul", "neg", "inv", "sin"]
15	"use_protected_ops" : True,
16	# input variables
17	"input_var_ids" : {"x" : 0 , "v" : 1 , "t" : 2, },
18	"input_var_units" : {"x" : [1, 0, 0] , "v" : [1, -1, 0] , "t" : [0, 1, 0] },
19	"input_var_complexity" : {"x" : 0. , "v" : 1. , "t" : 0., },
20	# constants
21	"constants" : {"pi" : np.pi , "M" : 1e6 , "const1" : 1 },
22	"constants_units" : {"pi" : [0, 0, 0] , "M" : [0, 0, 1] , "const1" : [0, 0, 0] },
23	"constants_complexity" : {"pi" : 0. , "M" : 1. , "const1" : 1. },
24	# free constants
25	"free_constants" : {"c" },
26	"free_constants_init_val" : {"c" : 10. },
27	"free_constants_units" : {"c" : [1, -1, 0] },
28	"free_constants_complexity" : {"c" : 0. },
29	}
30
31	my_lib = Lib.Library(args_make_tokens = args_make_tokens,	1✔
32	superparent_units = [2, -2, 1], superparent_name = "y")
33
34	# TEST PROGRAM
35	test_program_str = ["mul", "mul", "M", "n2", "c", "sub", "inv", "sqrt", "sub", "const1", "div", "n2", "v", "n2",	1✔
36	"c", "cos", "div", "sub", "div", "const1", "const1", "div", "v", "c", "div", "v", "c"]
37	test_program_idx = np.array([my_lib.lib_name_to_idx[tok_str] for tok_str in test_program_str])	1✔
38
39	# ------------------- EXPECTED LIVE BEHAVIOR -------------------
40	expected_constraints = np.array([	1✔
41	(True, [2., -2., 1., 0., 0., 0., 0.],),
42	(False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
43	(False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
44	(False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
45	(False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
46	(True, [0., 0., 0., 0., 0., 0., 0.],),
47	(True, [0., 0., 0., 0., 0., 0., 0.],),
48	(True, [-0., -0., -0., -0., -0., -0., -0.],),
49	(True, [-0., -0., -0., -0., -0., -0., -0.],),
50	(True, [-0., -0., -0., -0., -0., -0., -0.],),
51	(True, [0., 0., 0., 0., 0., 0., 0.],),
52	(False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
53	(False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
54	(True, [2., -2., 0., 0., 0., 0., 0.],),
55	(True, [1., -1., 0., 0., 0., 0., 0.],),
56	(True, [0., 0., 0., 0., 0., 0., 0.],),
57	(True, [0., 0., 0., 0., 0., 0., 0.],),
58	(False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
59	(False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
60	(False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
61	(False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
62	(True, [0., 0., 0., 0., 0., 0., 0.],),
63	(False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
64	(True, [1., -1., 0., 0., 0., 0., 0.],),
65	(True, [0., 0., 0., 0., 0., 0., 0.],),
66	(False, [np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN, np.NaN],),
67	(True, [1., -1., 0., 0., 0., 0., 0.],),
68	], dtype=object)
69	expected_tokens_cases_record = np.array([3, 70, 21, 72, 21, 73, 4, 5, 5, 4, 1, 70, 21, 73, 5, 1, 6, 70, 21, 21, 72,	1✔
70	20, 70, 73, 73, 70, 73])
71	expected_phy_units = np.array([x for x in expected_constraints[:, 1]], dtype=float)	1✔
72	expected_is_constraining = expected_constraints[:, 0].astype(bool)	1✔
73
74	# ------------------- EXPECTED UNITS AT FINAL STEP -------------------
75	expected_constraints_final = np.array([	1✔
76	(True, [ 2., -2., 1., 0., 0., 0., 0.],),
77	(True, [ 2., -2., 1., 0., 0., 0., 0.],),
78	(True, [ 0., 0., 1., 0., 0., 0., 0.],),
79	(True, [ 2., -2., 0., 0., 0., 0., 0.],),
80	(True, [ 1., -1., 0., 0., 0., 0., 0.],),
81	(True, [-0., -0., -0., -0., -0., -0., -0.],),
82	(True, [-0., -0., -0., -0., -0., -0., -0.],),
83	(True, [ 0., 0., 0., 0., 0., 0., 0.],),
84	(True, [ 0., 0., 0., 0., 0., 0., 0.],),
85	(True, [ 0., 0., 0., 0., 0., 0., 0.],),
86	(True, [ 0., 0., 0., 0., 0., 0., 0.],),
87	(True, [ 2., -2., 0., 0., 0., 0., 0.],),
88	(True, [ 1., -1., 0., 0., 0., 0., 0.],),
89	(True, [ 2., -2., 0., 0., 0., 0., 0.],),
90	(True, [ 1., -1., 0., 0., 0., 0., 0.],),
91	(True, [ 0., 0., 0., 0., 0., 0., 0.],),
92	(True, [ 0., 0., 0., 0., 0., 0., 0.],),
93	(True, [ 0., 0., 0., 0., 0., 0., 0.],),
94	(True, [ 0., 0., 0., 0., 0., 0., 0.],),
95	(True, [ 0., 0., 0., 0., 0., 0., 0.],),
96	(True, [ 0., 0., 0., 0., 0., 0., 0.],),
97	(True, [ 0., 0., 0., 0., 0., 0., 0.],),
98	(True, [ 1., -1., 0., 0., 0., 0., 0.],),
99	(True, [ 1., -1., 0., 0., 0., 0., 0.],),
100	(True, [ 0., 0., 0., 0., 0., 0., 0.],),
101	(True, [ 1., -1., 0., 0., 0., 0., 0.],),
102	(True, [ 1., -1., 0., 0., 0., 0., 0.],),
103	], dtype=object)
104
105	expected_phy_units_final = np.array([x for x in expected_constraints_final[:, 1]], dtype=float)	1✔
106	expected_is_constraining_final = expected_constraints_final[:, 0].astype(bool)	1✔
107
108	return (test_program_idx, my_lib, expected_tokens_cases_record, expected_phy_units, expected_is_constraining, \	1✔
109	expected_phy_units_final, expected_is_constraining_final)
110
111	class DimensionalAnalysisTest(unittest.TestCase):	1✔
112
113	def test_assign_units_bottom_up(self):	1✔
114	test_program_idx, my_lib, expected_tokens_cases_record, expected_phy_units, expected_is_constraining, \	1✔
115	expected_phy_units_final, expected_is_constraining_final = hard_test_case()
116
117	test_program_length = len(test_program_idx)	1✔
118	batch_size = 3	1✔
119	test_programs_idx = np.tile(test_program_idx, reps=(batch_size, 1))	1✔
120
121	my_programs = VProg.VectPrograms(batch_size=batch_size, max_time_step=test_program_length, library=my_lib, n_realizations=1)	1✔
122	my_programs.set_programs(test_programs_idx)	1✔
123
124	# ------------------- TEST BOTTOM-UP -------------------
125	try:	1✔
126	phy.assign_units_bottom_up(my_programs,	1✔
127	coords_start = my_programs.coords_of_step(0),
128	coords_end = my_programs.coords_of_step(my_programs.max_time_step-1),)
129	except:	×
130	self.fail("Bottom up run failed.")	×
131
132	# ------------------- TEST BOTTOM-UP RESULTS-------------------
133	# After bottom up
134	observed_phy_units_final_after_BU = my_programs.tokens.phy_units [0,:].copy()	1✔
135	observed_is_constraining_final_after_BU = my_programs.tokens.is_constraining_phy_units [0,:].copy()	1✔
136
137	# ASSERT BOTTOM UP GIVES RIGHT RESULT
138	bool_works = np.array_equal(expected_is_constraining_final , observed_is_constraining_final_after_BU)	1✔
139	self.assertEqual(bool_works, True)	1✔
140	bool_works = np.array_equal(expected_phy_units_final , observed_phy_units_final_after_BU, equal_nan = True )	1✔
141	self.assertEqual(bool_works, True)	1✔
142
143	def test_get_required_units(self):	1✔
144
145	# ------------------- TEST CASE -------------------
146	test_program_idx, my_lib, expected_tokens_cases_record, expected_phy_units, expected_is_constraining, \	1✔
147	expected_phy_units_final, expected_is_constraining_final = hard_test_case()
148
149	test_program_length = len(test_program_idx)	1✔
150	batch_size = 3	1✔
151	test_programs_idx = np.tile(test_program_idx, reps=(batch_size, 1)).transpose()	1✔
152
153	# ------------------- TRY AT STEP: assign_required_units_at_step -------------------
154	try:	1✔
155	batch_size_large = int(1e3)	1✔
156	test_programs_idx_large = np.tile(test_program_idx, reps=(batch_size_large, 1)).transpose()	1✔
157	my_programs = VProg.VectPrograms(batch_size=batch_size_large, max_time_step=test_program_length, library=my_lib, n_realizations=1)	1✔
158	t0 = time.perf_counter()	1✔
159	for i, idx in enumerate(test_programs_idx_large):	1✔
160	# Computing units requirements
161	cases_record = phy.assign_required_units_at_step(my_programs)	1✔
162	# Appending new token
163	# In real world run, next token should be chosen based on constraints found above
164	my_programs.append(idx)	1✔
165	t1 = time.perf_counter()	1✔
166	marginal_time = ((t1-t0)1e3)/(test_program_length batch_size_large)	1✔
167	print("\nRequired units time (in ideal non-mixed cases) : %f ms / step / (prog in batch) " % marginal_time)	1✔
168	except:	×
169	self.fail("Unable to run assign_required_units_at_step")	×
170
171	# ------------------- TEST AT STEP: assign_required_units_at_step -------------------
172	my_programs = VProg.VectPrograms(batch_size=batch_size, max_time_step=test_program_length, library=my_lib, n_realizations=1)	1✔
173
174	observed_phy_units = []	1✔
175	observed_is_constraining = []	1✔
176	observed_tokens_cases_record = []	1✔
177	for i, idx in enumerate(test_programs_idx) :	1✔
178	# Computing units requirements
179	cases_record = phy.assign_required_units_at_step(my_programs)	1✔
180	# Saving observed results
181	observed_tokens_cases_record. append (cases_record [0] .copy())	1✔
182	observed_phy_units. append (my_programs.tokens.phy_units [0,i].copy())	1✔
183	observed_is_constraining. append (my_programs.tokens.is_constraining_phy_units [0,i].copy())	1✔
184	# Assertions
185	bool_works = np.array_equal(observed_tokens_cases_record[i], expected_tokens_cases_record[i])	1✔
186	self.assertEqual(bool_works, True)	1✔
187	bool_works = np.array_equal(observed_is_constraining [i], expected_is_constraining [i])	1✔
188	self.assertEqual(bool_works, True)	1✔
189	bool_works = np.array_equal(observed_phy_units [i], expected_phy_units [i], equal_nan = True )	1✔
190	self.assertEqual(bool_works, True)	1✔
191	# Appending new token
192	# In real world run, next token should be chosen based on constraints found above
193	my_programs.append(idx)	1✔
194
195	# To numpy array
196	observed_phy_units = np.array(observed_phy_units)	1✔
197	observed_is_constraining = np.array(observed_is_constraining)	1✔
198	observed_tokens_cases_record = np.array(observed_tokens_cases_record)	1✔
199
200	# Final result after all idx were added (Results using assign_required_units_at_step function)
201	observed_phy_units_final_w_step = my_programs.tokens.phy_units [0,:].copy()	1✔
202	observed_is_constraining_final_w_step = my_programs.tokens.is_constraining_phy_units [0,:].copy()	1✔
203
204	# (Results using assign_required_units_at_step function)
205	observed_phy_units_w_step = observed_phy_units .copy()	1✔
206	observed_is_constraining_w_step = observed_is_constraining .copy()	1✔
207	observed_tokens_cases_record_w_step = observed_tokens_cases_record .copy()	1✔
208
209	# ------------------- TEST AT STEP: assign_required_units_at_step from scratch -------------------
210	my_programs = VProg.VectPrograms(batch_size=batch_size, max_time_step=test_program_length, library=my_lib, n_realizations=1)	1✔
211	my_programs.set_programs(test_programs_idx.transpose())	1✔
212
213	phy.assign_required_units_at_step(my_programs, from_scratch=True)	1✔
214
215	# Final result after all idx were added (Results using assign_required_units_at_step function)
216	observed_phy_units_final_w_step_from_scratch = my_programs.tokens.phy_units [0,:].copy()	1✔
217	observed_is_constraining_final_w_step_from_scratch = my_programs.tokens.is_constraining_phy_units [0,:].copy()	1✔
218
219	# Assert re-computing everything from_scratch gives same result
220	bool_works = np.array_equal(observed_phy_units_final_w_step_from_scratch , observed_phy_units_final_w_step , equal_nan=True)	1✔
221	self.assertEqual(bool_works, True)	1✔
222	bool_works = np.array_equal(observed_is_constraining_final_w_step_from_scratch , observed_is_constraining_final_w_step)	1✔
223	self.assertEqual(bool_works, True)	1✔
224
225	# ------------------- TEST AT COORDS: assign_required_units -------------------
226	my_programs = VProg.VectPrograms(batch_size=batch_size, max_time_step=test_program_length, library=my_lib, n_realizations=1)	1✔
227
228	observed_phy_units = []	1✔
229	observed_is_constraining = []	1✔
230	observed_tokens_cases_record = []	1✔
231	for i, idx in enumerate(test_programs_idx) :	1✔
232	# Computing units requirements
233	coords = my_programs.coords_of_step(i)[:, [0,2]] # doing 0-th and 2-th of batch but not 1-th	1✔
234	cases_record = phy.assign_required_units(my_programs, coords=coords)	1✔
235	# Saving observed results
236	observed_tokens_cases_record. append (cases_record [0] .copy())	1✔
237	observed_phy_units. append (my_programs.tokens.phy_units [0,i].copy())	1✔
238	observed_is_constraining. append (my_programs.tokens.is_constraining_phy_units [0,i].copy())	1✔
239	# Assertions
240	bool_works = np.array_equal(observed_tokens_cases_record[i], expected_tokens_cases_record[i])	1✔
241	self.assertEqual(bool_works, True)	1✔
242	bool_works = np.array_equal(observed_is_constraining [i], expected_is_constraining [i])	1✔
243	self.assertEqual(bool_works, True)	1✔
244	bool_works = np.array_equal(observed_phy_units [i], expected_phy_units [i], equal_nan = True )	1✔
245	self.assertEqual(bool_works, True)	1✔
246	# Appending new token
247	# In real world run, next token should be chosen based on constraints found above
248	my_programs.append(idx)	1✔
249
250	# To numpy array
251	observed_phy_units = np.array(observed_phy_units)	1✔
252	observed_is_constraining = np.array(observed_is_constraining)	1✔
253	observed_tokens_cases_record = np.array(observed_tokens_cases_record)	1✔
254
255	# Final result after all idx were added
256	observed_phy_units_final = my_programs.tokens.phy_units [0,:].copy()	1✔
257	observed_is_constraining_final = my_programs.tokens.is_constraining_phy_units [0,:].copy()	1✔
258
259	# Assert that assign_required_units at coords gives same results as assign_required_units_at_step
260	# Final result
261	bool_works = np.array_equal(observed_phy_units_final , observed_phy_units_final_w_step , equal_nan=True)	1✔
262	self.assertEqual(bool_works, True)	1✔
263	bool_works = np.array_equal(observed_is_constraining_final , observed_is_constraining_final_w_step)	1✔
264	self.assertEqual(bool_works, True)	1✔
265	# Intermediate results
266	bool_works = np.array_equal(observed_phy_units , observed_phy_units_w_step , equal_nan=True)	1✔
267	self.assertEqual(bool_works, True)	1✔
268	bool_works = np.array_equal(observed_is_constraining , observed_is_constraining_w_step)	1✔
269	self.assertEqual(bool_works, True)	1✔
270	bool_works = np.array_equal(observed_tokens_cases_record , observed_tokens_cases_record_w_step)	1✔
271	self.assertEqual(bool_works, True)	1✔
272
273	# ------------------- CHECK ALL THE UNITS FOUND ARE COHERENT WITH BOTTOM-UP -------------------
274	try:	1✔
275	phy.assign_units_bottom_up(my_programs,	1✔
276	coords_start = my_programs.coords_of_step(0),
277	coords_end = my_programs.coords_of_step(my_programs.max_time_step-1),)
278	except:	×
279	self.fail("Final check bottom up failed.")	×
280
281	# After bottom up
282	observed_phy_units_final_after_BU = my_programs.tokens.phy_units [0,:].copy()	1✔
283	observed_is_constraining_final_after_BU = my_programs.tokens.is_constraining_phy_units [0,:].copy()	1✔
284
285	# ASSERT BOTTOM UP GIVES RIGHT RESULT
286	bool_works = np.array_equal(expected_is_constraining_final , observed_is_constraining_final_after_BU)	1✔
287	self.assertEqual(bool_works, True)	1✔
288	bool_works = np.array_equal(expected_phy_units_final , observed_phy_units_final_after_BU, equal_nan = True )	1✔
289	self.assertEqual(bool_works, True)	1✔
290
291	# ASSERT THAT END STEP OF LIVE UNITS CONSTRAINTS GIVES SAME AS BOTTOM UP
292	bool_works = np.array_equal(observed_is_constraining_final , observed_is_constraining_final_after_BU)	1✔
293	self.assertEqual(bool_works, True)	1✔
294	bool_works = np.array_equal(observed_phy_units_final , observed_phy_units_final_after_BU, equal_nan = True )	1✔
295	self.assertEqual(bool_works, True)	1✔
296
297	return None	1✔
298
299
300	if __name__ == '__main__':	1✔
301	unittest.main(verbosity=2)	×

WassimTenachi / PhySO / #13

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous