<s> __author__ = '<STR_LIT>' <EOL> import os <EOL> import platform <EOL> import logging <EOL> import time <EOL> import numpy as np <EOL> from pypet . trajectory import Trajectory , load_trajectory <EOL> from pypet . utils . explore import cartesian_product <EOL> from pypet . environment import Environment <EOL> from pypet . storageservice import HDF5StorageService <EOL> from pypet import pypetconstants , Result , manual_run <EOL> import pypet . pypetexceptions as pex <EOL> import sys <EOL> if ( sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:7> , <NUM_LIT:0> ) ) : <EOL> import unittest2 as unittest <EOL> else : <EOL> import unittest <EOL> try : <EOL> import psutil <EOL> except ImportError : <EOL> psutil = None <EOL> try : <EOL> import dill <EOL> except ImportError : <EOL> dill = None <EOL> import scipy . sparse as spsp <EOL> import random <EOL> import pypet . compat as compat <EOL> import pypet . utils . ptcompat as ptcompat <EOL> from pypet import Parameter <EOL> import tables as pt <EOL> from pypet . tests . testutils . ioutils import run_suite , make_temp_dir , make_trajectory_name , get_root_logger , parse_args , get_log_config , get_log_path <EOL> from pypet . tests . testutils . data import create_param_dict , add_params , multiply , simple_calculations , TrajectoryComparator , multiply_args , multiply_with_storing , multiply_with_graceful_exit <EOL> def add_one_particular_item ( traj , store_full ) : <EOL> traj . hi = Result ( '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' ) <EOL> traj . f_store ( ) <EOL> traj . f_remove_child ( '<STR_LIT>' ) <EOL> class SlowResult ( Result ) : <EOL> def _load ( self , load_dict ) : <EOL> time . sleep ( <NUM_LIT:3> ) <EOL> super ( SlowResult , self ) . _load ( load_dict ) <EOL> class FullStorageTest ( TrajectoryComparator ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> def test_full_store ( self ) : <EOL> filename = make_temp_dir ( '<STR_LIT>' ) <EOL> with Environment ( filename = filename , <EOL> log_config = get_log_config ( ) ) as env : <EOL> traj = env . v_trajectory <EOL> traj . par . x = Parameter ( '<STR_LIT:x>' , <NUM_LIT:3> , '<STR_LIT>' ) <EOL> traj . f_explore ( { '<STR_LIT:x>' : [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] } ) <EOL> env . f_run ( add_one_particular_item , True ) <EOL> traj = load_trajectory ( index = - <NUM_LIT:1> , filename = filename ) <EOL> self . assertTrue ( '<STR_LIT>' in traj ) <EOL> def with_niceness ( traj ) : <EOL> if traj . multiproc : <EOL> if hasattr ( os , '<STR_LIT>' ) : <EOL> trajnice = traj . niceness <EOL> osnice = os . nice ( <NUM_LIT:0> ) <EOL> else : <EOL> trajnice = traj . niceness <EOL> osnice = psutil . Process ( ) . nice ( ) <EOL> if trajnice != osnice : <EOL> if traj . use_scoop : <EOL> import scoop <EOL> if ( not scoop . IS_RUNNING or scoop . IS_ORIGIN ) : <EOL> return <EOL> raise RuntimeError ( '<STR_LIT>' <EOL> '<STR_LIT>' % ( str ( trajnice ) , str ( osnice ) ) ) <EOL> def add_large_data ( traj ) : <EOL> np_array = np . random . rand ( <NUM_LIT:100> , <NUM_LIT:1000> , <NUM_LIT:10> ) <EOL> traj . f_add_result ( '<STR_LIT>' , np_array ) <EOL> traj . f_store_item ( '<STR_LIT>' ) <EOL> traj . f_remove_item ( '<STR_LIT>' ) <EOL> array_list = [ ] <EOL> for irun in range ( <NUM_LIT> ) : <EOL> array_list . append ( np . random . rand ( <NUM_LIT:10> ) ) <EOL> traj . f_add_result ( '<STR_LIT>' , * array_list ) <EOL> class SimpleEnvironmentTest ( TrajectoryComparator ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> def test_make_default_file_when_giving_directory_without_slash ( self ) : <EOL> filename = make_temp_dir ( '<STR_LIT>' ) <EOL> head , tail = os . path . split ( filename ) <EOL> env = Environment ( filename = head ) <EOL> the_file_name = env . v_traj . v_name + '<STR_LIT>' <EOL> head , tail = os . path . split ( env . v_traj . v_storage_service . filename ) <EOL> self . assertEqual ( tail , the_file_name ) <EOL> class EnvironmentTest ( TrajectoryComparator ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> def set_mode ( self ) : <EOL> self . mode = '<STR_LIT>' <EOL> self . multiproc = False <EOL> self . gc_interval = None <EOL> self . ncores = <NUM_LIT:1> <EOL> self . use_pool = True <EOL> self . use_scoop = False <EOL> self . freeze_input = False <EOL> self . pandas_format = '<STR_LIT>' <EOL> self . pandas_append = False <EOL> self . complib = '<STR_LIT>' <EOL> self . complevel = <NUM_LIT:9> <EOL> self . shuffle = True <EOL> self . fletcher32 = False <EOL> self . encoding = '<STR_LIT:utf8>' <EOL> self . log_stdout = False <EOL> self . wildcard_functions = None <EOL> self . niceness = None <EOL> self . port = None <EOL> self . timeout = None <EOL> self . add_time = True <EOL> self . graceful_exit = False <EOL> def explore_complex_params ( self , traj ) : <EOL> matrices_csr = [ ] <EOL> for irun in range ( <NUM_LIT:3> ) : <EOL> spsparse_csr = spsp . lil_matrix ( ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> spsparse_csr [ <NUM_LIT:3> , <NUM_LIT:2> + irun ] = <NUM_LIT> * irun <EOL> matrices_csr . append ( spsparse_csr . tocsr ( ) ) <EOL> matrices_csc = [ ] <EOL> for irun in range ( <NUM_LIT:3> ) : <EOL> spsparse_csc = spsp . lil_matrix ( ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> spsparse_csc [ <NUM_LIT:3> , <NUM_LIT:2> + irun ] = <NUM_LIT> * irun <EOL> matrices_csc . append ( spsparse_csc . tocsc ( ) ) <EOL> matrices_bsr = [ ] <EOL> for irun in range ( <NUM_LIT:3> ) : <EOL> spsparse_bsr = spsp . lil_matrix ( ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> spsparse_bsr [ <NUM_LIT:3> , <NUM_LIT:2> + irun ] = <NUM_LIT> * irun <EOL> matrices_bsr . append ( spsparse_bsr . tocsr ( ) . tobsr ( ) ) <EOL> matrices_dia = [ ] <EOL> for irun in range ( <NUM_LIT:3> ) : <EOL> spsparse_dia = spsp . lil_matrix ( ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> spsparse_dia [ <NUM_LIT:3> , <NUM_LIT:2> + irun ] = <NUM_LIT> * irun <EOL> matrices_dia . append ( spsparse_dia . tocsc ( ) . todia ( ) ) <EOL> self . explore_dict = { '<STR_LIT:string>' : [ np . array ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> np . array ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> np . array ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) ] , <EOL> '<STR_LIT:int>' : [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , <EOL> '<STR_LIT>' : matrices_csr , <EOL> '<STR_LIT>' : matrices_csc , <EOL> '<STR_LIT>' : matrices_bsr , <EOL> '<STR_LIT>' : matrices_dia , <EOL> '<STR_LIT:list>' : [ [ '<STR_LIT>' ] , [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:4> , <NUM_LIT:4> , <NUM_LIT:4> ] , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT> ] ] } <EOL> with self . assertRaises ( pex . NotUniqueNodeError ) : <EOL> traj . f_explore ( self . explore_dict ) <EOL> traj . f_shrink ( force = True ) <EOL> par_dict = traj . parameters . f_to_dict ( ) <EOL> for param_name in par_dict : <EOL> param = par_dict [ param_name ] <EOL> if param . v_name in self . explore_dict : <EOL> param . f_unlock ( ) <EOL> if param . v_explored : <EOL> param . _shrink ( ) <EOL> self . explore_dict = { '<STR_LIT>' : [ np . array ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> np . array ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> np . array ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) ] , <EOL> '<STR_LIT>' : [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , <EOL> '<STR_LIT>' : matrices_csr , <EOL> '<STR_LIT>' : matrices_csc , <EOL> '<STR_LIT>' : matrices_bsr , <EOL> '<STR_LIT>' : matrices_dia , <EOL> '<STR_LIT:list>' : [ [ '<STR_LIT>' ] , [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:4> , <NUM_LIT:4> , <NUM_LIT:4> ] , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT> ] ] } <EOL> traj . f_explore ( self . explore_dict ) <EOL> def explore ( self , traj ) : <EOL> self . explored = { '<STR_LIT>' : [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : [ np . array ( [ <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) , np . array ( [ - <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) ] , <EOL> '<STR_LIT>' : [ spsp . lil_matrix ( ( <NUM_LIT> , <NUM_LIT> ) ) , spsp . lil_matrix ( ( <NUM_LIT> , <NUM_LIT> ) ) ] } <EOL> self . explored [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ <NUM_LIT:1> , <NUM_LIT:2> ] = <NUM_LIT> <EOL> self . explored [ '<STR_LIT>' ] [ <NUM_LIT:1> ] [ <NUM_LIT:2> , <NUM_LIT:2> ] = <NUM_LIT> <EOL> self . explored [ '<STR_LIT>' ] [ <NUM_LIT:0> ] = self . explored [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . tocsr ( ) <EOL> self . explored [ '<STR_LIT>' ] [ <NUM_LIT:1> ] = self . explored [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . tocsr ( ) <EOL> traj . f_explore ( cartesian_product ( self . explored ) ) <EOL> def explore_large ( self , traj ) : <EOL> self . explored = { '<STR_LIT>' : [ <NUM_LIT:0> , <NUM_LIT:1> ] } <EOL> traj . f_explore ( cartesian_product ( self . explored ) ) <EOL> def tearDown ( self ) : <EOL> self . env . f_disable_logging ( ) <EOL> super ( EnvironmentTest , self ) . tearDown ( ) <EOL> def setUp ( self ) : <EOL> self . set_mode ( ) <EOL> self . logfolder = make_temp_dir ( os . path . join ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) ) <EOL> random . seed ( ) <EOL> self . trajname = make_trajectory_name ( self ) <EOL> self . filename = make_temp_dir ( os . path . join ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' % self . trajname ) ) <EOL> env = Environment ( trajectory = self . trajname , filename = self . filename , <EOL> file_title = self . trajname , <EOL> log_stdout = self . log_stdout , <EOL> log_config = get_log_config ( ) , <EOL> results_per_run = <NUM_LIT:5> , <EOL> wildcard_functions = self . wildcard_functions , <EOL> derived_parameters_per_run = <NUM_LIT:5> , <EOL> multiproc = self . multiproc , <EOL> ncores = self . ncores , <EOL> wrap_mode = self . mode , <EOL> use_pool = self . use_pool , <EOL> gc_interval = self . gc_interval , <EOL> freeze_input = self . freeze_input , <EOL> fletcher32 = self . fletcher32 , <EOL> complevel = self . complevel , <EOL> complib = self . complib , <EOL> shuffle = self . shuffle , <EOL> pandas_append = self . pandas_append , <EOL> pandas_format = self . pandas_format , <EOL> encoding = self . encoding , <EOL> niceness = self . niceness , <EOL> use_scoop = self . use_scoop , <EOL> port = self . port , <EOL> add_time = self . add_time , <EOL> timeout = self . timeout , <EOL> graceful_exit = self . graceful_exit ) <EOL> traj = env . v_trajectory <EOL> traj . v_standard_parameter = Parameter <EOL> self . param_dict = { } <EOL> create_param_dict ( self . param_dict ) <EOL> add_params ( traj , self . param_dict ) <EOL> self . traj = traj <EOL> self . env = env <EOL> @ unittest . skipIf ( not hasattr ( os , '<STR_LIT>' ) and psutil is None , '<STR_LIT>' ) <EOL> def test_niceness ( self ) : <EOL> self . explore ( self . traj ) <EOL> self . env . f_run ( with_niceness ) <EOL> self . assertTrue ( self . traj . f_is_completed ( ) ) <EOL> def test_file_overwriting ( self ) : <EOL> self . traj . f_store ( ) <EOL> with ptcompat . open_file ( self . filename , mode = '<STR_LIT:r>' ) as file : <EOL> nchildren = len ( file . root . _v_children ) <EOL> self . assertTrue ( nchildren > <NUM_LIT:0> ) <EOL> env2 = Environment ( filename = self . filename , <EOL> log_config = get_log_config ( ) ) <EOL> traj2 = env2 . v_trajectory <EOL> traj2 . f_store ( ) <EOL> self . assertTrue ( os . path . exists ( self . filename ) ) <EOL> with ptcompat . open_file ( self . filename , mode = '<STR_LIT:r>' ) as file : <EOL> nchildren = len ( file . root . _v_children ) <EOL> self . assertTrue ( nchildren > <NUM_LIT:1> ) <EOL> env3 = Environment ( filename = self . filename , overwrite_file = True , <EOL> log_config = get_log_config ( ) ) <EOL> self . assertFalse ( os . path . exists ( self . filename ) ) <EOL> env2 . f_disable_logging ( ) <EOL> env3 . f_disable_logging ( ) <EOL> def test_time_display_of_loading ( self ) : <EOL> filename = make_temp_dir ( '<STR_LIT>' ) <EOL> env = Environment ( trajectory = '<STR_LIT>' , add_time = True , filename = filename , <EOL> log_stdout = False , <EOL> log_config = get_log_config ( ) , <EOL> dynamic_imports = SlowResult , <EOL> display_time = <NUM_LIT:0.1> ) <EOL> traj = env . v_traj <EOL> res = traj . f_add_result ( SlowResult , '<STR_LIT>' , <NUM_LIT> , <NUM_LIT> , comment = '<STR_LIT>' ) <EOL> traj . f_store ( ) <EOL> service_logger = traj . v_storage_service . _logger <EOL> root = logging . getLogger ( '<STR_LIT>' ) <EOL> old_level = root . level <EOL> service_logger . setLevel ( logging . INFO ) <EOL> root . setLevel ( logging . INFO ) <EOL> traj . f_load ( load_data = <NUM_LIT:3> ) <EOL> service_logger . setLevel ( old_level ) <EOL> root . setLevel ( old_level ) <EOL> path = get_log_path ( traj ) <EOL> mainfilename = os . path . join ( path , '<STR_LIT>' ) <EOL> with open ( mainfilename , mode = '<STR_LIT:r>' ) as mainf : <EOL> full_text = mainf . read ( ) <EOL> self . assertTrue ( '<STR_LIT>' in full_text ) <EOL> env . f_disable_logging ( ) <EOL> def make_run_large_data ( self ) : <EOL> self . env . f_run ( add_large_data ) <EOL> def make_run ( self ) : <EOL> simple_arg = - <NUM_LIT> <EOL> simple_kwarg = <NUM_LIT> <EOL> results = self . env . f_run ( simple_calculations , simple_arg , simple_kwarg = simple_kwarg ) <EOL> self . are_results_in_order ( results ) <EOL> def test_a_large_run ( self ) : <EOL> get_root_logger ( ) . info ( '<STR_LIT>' ) <EOL> self . traj . f_add_parameter ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . explore_large ( self . traj ) <EOL> self . make_run_large_data ( ) <EOL> self . assertTrue ( self . traj . f_is_completed ( ) ) <EOL> get_root_logger ( ) . info ( str ( self . env ) ) <EOL> get_root_logger ( ) . info ( repr ( self . env ) ) <EOL> newtraj = Trajectory ( ) <EOL> newtraj . f_load ( name = self . traj . v_name , as_new = False , load_data = <NUM_LIT:2> , filename = self . filename ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . compare_trajectories ( self . traj , newtraj ) <EOL> size = os . path . getsize ( self . filename ) <EOL> size_in_mb = size / <NUM_LIT> <EOL> get_root_logger ( ) . info ( '<STR_LIT>' % str ( size_in_mb ) ) <EOL> self . assertTrue ( size_in_mb < <NUM_LIT> , '<STR_LIT>' % str ( size_in_mb ) ) <EOL> def test_two_runs ( self ) : <EOL> self . traj . f_add_parameter ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . traj . hdf5 . purge_duplicate_comments = False <EOL> self . explore ( self . traj ) <EOL> self . make_run ( ) <EOL> newtraj = self . load_trajectory ( trajectory_name = self . traj . v_name , as_new = False ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . compare_trajectories ( self . traj , newtraj ) <EOL> size = os . path . getsize ( self . filename ) <EOL> size_in_mb = size / <NUM_LIT> <EOL> get_root_logger ( ) . info ( '<STR_LIT>' % str ( size_in_mb ) ) <EOL> self . assertTrue ( size_in_mb < <NUM_LIT> , '<STR_LIT>' % str ( size_in_mb ) ) <EOL> mp_traj = self . traj <EOL> old_multiproc = self . multiproc <EOL> self . multiproc = False <EOL> self . setUp ( ) <EOL> self . traj . f_add_parameter ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . traj . hdf5 . purge_duplicate_comments = False <EOL> self . explore ( self . traj ) <EOL> self . make_run ( ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . compare_trajectories ( self . traj , newtraj ) <EOL> size = os . path . getsize ( self . filename ) <EOL> size_in_mb = size / <NUM_LIT> <EOL> get_root_logger ( ) . info ( '<STR_LIT>' % str ( size_in_mb ) ) <EOL> self . assertTrue ( size_in_mb < <NUM_LIT> , '<STR_LIT>' % str ( size_in_mb ) ) <EOL> self . compare_trajectories ( mp_traj , self . traj ) <EOL> self . multiproc = old_multiproc <EOL> def test_errors ( self ) : <EOL> tmp = make_temp_dir ( '<STR_LIT>' ) <EOL> if dill is not None : <EOL> env1 = Environment ( continuable = True , continue_folder = tmp , <EOL> log_config = None , filename = self . filename ) <EOL> with self . assertRaises ( ValueError ) : <EOL> env1 . f_run_map ( multiply_args , [ <NUM_LIT:1> ] , [ <NUM_LIT:2> ] , [ <NUM_LIT:3> ] ) <EOL> with self . assertRaises ( ValueError ) : <EOL> Environment ( multiproc = True , use_pool = False , freeze_input = True , <EOL> filename = self . filename , log_config = None ) <EOL> env3 = Environment ( log_config = None , filename = self . filename ) <EOL> with self . assertRaises ( ValueError ) : <EOL> env3 . f_run_map ( multiply_args ) <EOL> with self . assertRaises ( ValueError ) : <EOL> Environment ( use_scoop = True , immediate_postproc = True ) <EOL> with self . assertRaises ( ValueError ) : <EOL> Environment ( use_pool = True , immediate_postproc = True ) <EOL> with self . assertRaises ( ValueError ) : <EOL> Environment ( continuable = True , wrap_mode = '<STR_LIT>' , continue_folder = tmp ) <EOL> with self . assertRaises ( ValueError ) : <EOL> Environment ( use_scoop = True , wrap_mode = '<STR_LIT>' ) <EOL> with self . assertRaises ( ValueError ) : <EOL> Environment ( automatic_storing = False , <EOL> continuable = True , continue_folder = tmp ) <EOL> with self . assertRaises ( ValueError ) : <EOL> Environment ( port = '<STR_LIT>' , wrap_mode = '<STR_LIT>' ) <EOL> def test_run ( self ) : <EOL> self . traj . f_add_parameter ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . explore ( self . traj ) <EOL> self . make_run ( ) <EOL> self . assertTrue ( self . traj . f_is_completed ( ) ) <EOL> newtraj = self . load_trajectory ( trajectory_name = self . traj . v_name , as_new = False ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . compare_trajectories ( self . traj , newtraj ) <EOL> size = os . path . getsize ( self . filename ) <EOL> size_in_mb = size / <NUM_LIT> <EOL> get_root_logger ( ) . info ( '<STR_LIT>' % str ( size_in_mb ) ) <EOL> self . assertTrue ( size_in_mb < <NUM_LIT> , '<STR_LIT>' % str ( size_in_mb ) ) <EOL> def test_just_one_run ( self ) : <EOL> self . make_run ( ) <EOL> self . assertTrue ( self . traj . f_is_completed ( ) ) <EOL> newtraj = self . load_trajectory ( trajectory_name = self . traj . v_name , as_new = False ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . compare_trajectories ( self . traj , newtraj ) <EOL> self . assertTrue ( len ( newtraj ) == <NUM_LIT:1> ) <EOL> size = os . path . getsize ( self . filename ) <EOL> size_in_mb = size / <NUM_LIT> <EOL> get_root_logger ( ) . info ( '<STR_LIT>' % str ( size_in_mb ) ) <EOL> self . assertTrue ( size_in_mb < <NUM_LIT> , '<STR_LIT>' % str ( size_in_mb ) ) <EOL> with self . assertRaises ( TypeError ) : <EOL> self . explore ( self . traj ) <EOL> def test_run_complex ( self ) : <EOL> self . traj . f_add_parameter ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . explore_complex_params ( self . traj ) <EOL> self . make_run ( ) <EOL> self . assertTrue ( self . traj . f_is_completed ( ) ) <EOL> newtraj = self . load_trajectory ( trajectory_name = self . traj . v_name , as_new = False ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . compare_trajectories ( self . traj , newtraj ) <EOL> def load_trajectory ( self , trajectory_index = None , trajectory_name = None , as_new = False ) : <EOL> newtraj = Trajectory ( ) <EOL> newtraj . v_storage_service = HDF5StorageService ( filename = self . filename ) <EOL> newtraj . f_load ( name = trajectory_name , index = trajectory_index , as_new = as_new , <EOL> load_parameters = <NUM_LIT:2> , load_derived_parameters = <NUM_LIT:2> , load_results = <NUM_LIT:2> , <EOL> load_other_data = <NUM_LIT:2> ) <EOL> return newtraj <EOL> def test_expand ( self ) : <EOL> self . traj . f_add_parameter ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . explore ( self . traj ) <EOL> self . make_run ( ) <EOL> self . expand ( ) <EOL> get_root_logger ( ) . info ( '<STR_LIT>' ) <EOL> self . make_run ( ) <EOL> newtraj = self . load_trajectory ( trajectory_name = self . traj . v_name , as_new = False ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . compare_trajectories ( self . traj , newtraj ) <EOL> def test_expand_after_reload ( self ) : <EOL> self . traj . f_add_parameter ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . explore ( self . traj ) <EOL> self . make_run ( ) <EOL> traj_name = self . traj . v_name <EOL> self . env = Environment ( trajectory = self . traj , <EOL> log_stdout = False , <EOL> log_config = get_log_config ( ) ) <EOL> self . traj = self . env . v_trajectory <EOL> self . traj . f_load ( name = traj_name ) <EOL> self . traj . res . f_remove ( ) <EOL> self . traj . dpar . f_remove ( ) <EOL> self . expand ( ) <EOL> get_root_logger ( ) . info ( '<STR_LIT>' ) <EOL> self . make_run ( ) <EOL> newtraj = self . load_trajectory ( trajectory_name = self . traj . v_name , as_new = False ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . compare_trajectories ( self . traj , newtraj ) <EOL> def expand ( self ) : <EOL> self . expanded = { '<STR_LIT>' : [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : [ np . array ( [ <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) , np . array ( [ - <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) ] , <EOL> '<STR_LIT>' : [ spsp . lil_matrix ( ( <NUM_LIT> , <NUM_LIT> ) ) , spsp . lil_matrix ( ( <NUM_LIT> , <NUM_LIT> ) ) ] } <EOL> self . expanded [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ <NUM_LIT:1> , <NUM_LIT:2> ] = <NUM_LIT> <EOL> self . expanded [ '<STR_LIT>' ] [ <NUM_LIT:1> ] [ <NUM_LIT:2> , <NUM_LIT:2> ] = <NUM_LIT> <EOL> self . expanded [ '<STR_LIT>' ] [ <NUM_LIT:0> ] = self . expanded [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . tocsr ( ) <EOL> self . expanded [ '<STR_LIT>' ] [ <NUM_LIT:1> ] = self . expanded [ '<STR_LIT>' ] [ <NUM_LIT:1> ] . tocsr ( ) <EOL> self . traj . f_expand ( cartesian_product ( self . expanded ) ) <EOL> self . traj . f_store ( ) <EOL> def test_switch_ON_large_tables ( self ) : <EOL> self . traj . f_add_parameter ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . explore ( self . traj ) <EOL> self . env . _traj . config . hdf5 . overview . results_overview = <NUM_LIT:1> <EOL> self . env . _traj . config . hdf5 . overview . derived_parameters_overview = <NUM_LIT:1> <EOL> self . make_run ( ) <EOL> hdf5file = pt . openFile ( self . filename ) <EOL> overview_group = hdf5file . getNode ( where = '<STR_LIT:/>' + self . traj . v_name , name = '<STR_LIT>' ) <EOL> should = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> for name in should : <EOL> self . assertTrue ( name in overview_group , '<STR_LIT>' % name ) <EOL> hdf5file . close ( ) <EOL> self . traj . f_load ( load_parameters = <NUM_LIT:2> , load_derived_parameters = <NUM_LIT:2> , load_results = <NUM_LIT:2> ) <EOL> newtraj = self . load_trajectory ( trajectory_name = self . traj . v_name ) <EOL> self . compare_trajectories ( newtraj , self . traj ) <EOL> def test_switch_off_all_tables ( self ) : <EOL> self . traj . f_add_parameter ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . explore ( self . traj ) <EOL> self . env . _traj . config . hdf5 . overview . results_overview = <NUM_LIT:0> <EOL> self . env . _traj . config . hdf5 . overview . derived_parameters_overview = <NUM_LIT:0> <EOL> self . env . _traj . config . hdf5 . overview . derived_parameters_summary = <NUM_LIT:0> <EOL> self . env . _traj . config . hdf5 . overview . results_summary = <NUM_LIT:0> <EOL> self . env . _traj . config . hdf5 . purge_duplicate_comments = <NUM_LIT:0> <EOL> self . env . _traj . config . hdf5 . overview . parameters_overview = <NUM_LIT:0> <EOL> self . env . _traj . config . hdf5 . overview . config_overview = <NUM_LIT:0> <EOL> self . env . _traj . config . hdf5 . overview . explored_parameters_overview = <NUM_LIT:0> <EOL> self . make_run ( ) <EOL> hdf5file = pt . openFile ( self . filename ) <EOL> overview_group = hdf5file . getNode ( where = '<STR_LIT:/>' + self . traj . v_name , name = '<STR_LIT>' ) <EOL> should_not = HDF5StorageService . NAME_TABLE_MAPPING . keys ( ) <EOL> for name in should_not : <EOL> name = name . split ( '<STR_LIT:.>' ) [ - <NUM_LIT:1> ] <EOL> self . assertTrue ( not name in overview_group , '<STR_LIT>' % name ) <EOL> hdf5file . close ( ) <EOL> def test_store_form_tuple ( self ) : <EOL> self . traj . f_store ( ) <EOL> self . traj . f_add_result ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT> ) <EOL> with self . assertRaises ( ValueError ) : <EOL> self . traj . f_store_item ( ( pypetconstants . LEAF , self . traj . TestResItem , ( ) , { } , <NUM_LIT:5> ) ) <EOL> self . traj . f_store_item ( ( pypetconstants . LEAF , self . traj . TestResItem ) ) <EOL> self . traj . results . f_remove_child ( '<STR_LIT>' ) <EOL> self . assertTrue ( '<STR_LIT>' not in self . traj ) <EOL> self . traj . results . f_load_child ( '<STR_LIT>' , load_data = pypetconstants . LOAD_SKELETON ) <EOL> self . traj . f_load_item ( ( pypetconstants . LEAF , self . traj . TestResItem , ( ) , { '<STR_LIT>' : '<STR_LIT>' } ) ) <EOL> self . assertTrue ( self . traj . TestResItem , <NUM_LIT> ) <EOL> def test_store_single_group ( self ) : <EOL> self . traj . f_store ( ) <EOL> self . traj . f_add_parameter_group ( '<STR_LIT>' ) . v_annotations . f_set ( <NUM_LIT> ) <EOL> self . traj . f_store_item ( '<STR_LIT>' ) <EOL> with self . assertRaises ( ValueError ) : <EOL> self . traj . parameters . new . f_remove_child ( '<STR_LIT>' ) <EOL> with self . assertRaises ( ValueError ) : <EOL> self . traj . parameters . new . f_store_child ( '<STR_LIT>' ) <EOL> with self . assertRaises ( TypeError ) : <EOL> self . traj . parameters . new . f_remove_child ( '<STR_LIT:test>' ) <EOL> self . traj . new . f_remove_child ( '<STR_LIT:test>' , recursive = True ) <EOL> self . assertTrue ( '<STR_LIT>' not in self . traj ) <EOL> self . traj . new . f_load_child ( '<STR_LIT:test>' , recursive = True , load_data = pypetconstants . LOAD_SKELETON ) <EOL> self . assertTrue ( self . traj . new . group . v_annotations . annotation , <NUM_LIT> ) <EOL> self . traj . f_delete_item ( '<STR_LIT>' ) <EOL> with self . assertRaises ( pex . DataNotInStorageError ) : <EOL> self . traj . parameters . f_load_child ( '<STR_LIT>' , <EOL> load_data = pypetconstants . LOAD_SKELETON ) <EOL> def test_switch_on_all_comments ( self ) : <EOL> self . explore ( self . traj ) <EOL> self . traj . hdf5 . purge_duplicate_comments = <NUM_LIT:0> <EOL> self . make_run ( ) <EOL> hdf5file = pt . openFile ( self . filename ) <EOL> traj_group = hdf5file . getNode ( where = '<STR_LIT:/>' , name = self . traj . v_name ) <EOL> for node in traj_group . _f_walkGroups ( ) : <EOL> if '<STR_LIT>' in node . _v_attrs : <EOL> self . assertTrue ( '<STR_LIT>' in node . _v_attrs , <EOL> '<STR_LIT>' % node . _v_name ) <EOL> hdf5file . close ( ) <EOL> def test_purge_duplicate_comments ( self ) : <EOL> self . explore ( self . traj ) <EOL> with self . assertRaises ( RuntimeError ) : <EOL> self . traj . hdf5 . purge_duplicate_comments = <NUM_LIT:1> <EOL> self . traj . overview . results_summary = <NUM_LIT:0> <EOL> self . make_run ( ) <EOL> self . traj . f_get ( '<STR_LIT>' ) . f_unlock ( ) <EOL> self . traj . hdf5 . purge_duplicate_comments = <NUM_LIT:1> <EOL> self . traj . f_get ( '<STR_LIT>' ) . f_unlock ( ) <EOL> self . traj . overview . results_summary = <NUM_LIT:1> <EOL> self . make_run ( ) <EOL> hdf5file = pt . openFile ( self . filename , mode = '<STR_LIT:a>' ) <EOL> ncomments = { } <EOL> try : <EOL> traj_group = hdf5file . getNode ( where = '<STR_LIT:/>' , name = self . traj . v_name ) <EOL> for node in traj_group . _f_walkGroups ( ) : <EOL> if ( '<STR_LIT>' in node . _v_pathname or <EOL> '<STR_LIT>' in node . _v_pathname ) : <EOL> if '<STR_LIT>' in node . _v_attrs : <EOL> if '<STR_LIT>' in node . _v_attrs : <EOL> comment = node . _v_attrs [ '<STR_LIT>' ] <EOL> if comment not in ncomments : <EOL> ncomments [ comment ] = <NUM_LIT:0> <EOL> ncomments [ comment ] += <NUM_LIT:1> <EOL> finally : <EOL> hdf5file . close ( ) <EOL> self . assertGreaterEqual ( len ( ncomments ) , <NUM_LIT:1> ) <EOL> self . assertTrue ( all ( x == <NUM_LIT:1> for x in ncomments . values ( ) ) ) <EOL> def test_NOT_purge_duplicate_comments ( self ) : <EOL> self . explore ( self . traj ) <EOL> self . traj . f_get ( '<STR_LIT>' ) . f_unlock ( ) <EOL> self . traj . hdf5 . purge_duplicate_comments = <NUM_LIT:0> <EOL> self . traj . f_get ( '<STR_LIT>' ) . f_unlock ( ) <EOL> self . traj . overview . results_summary = <NUM_LIT:0> <EOL> self . make_run ( ) <EOL> hdf5file = pt . openFile ( self . filename , mode = '<STR_LIT:a>' ) <EOL> ncomments = { } <EOL> try : <EOL> traj_group = hdf5file . getNode ( where = '<STR_LIT:/>' , name = self . traj . v_name ) <EOL> for node in traj_group . _f_walkGroups ( ) : <EOL> if ( '<STR_LIT>' in node . _v_pathname or <EOL> '<STR_LIT>' in node . _v_pathname ) : <EOL> if '<STR_LIT>' in node . _v_attrs : <EOL> if '<STR_LIT>' in node . _v_attrs : <EOL> comment = node . _v_attrs [ '<STR_LIT>' ] <EOL> if comment not in ncomments : <EOL> ncomments [ comment ] = <NUM_LIT:0> <EOL> ncomments [ comment ] += <NUM_LIT:1> <EOL> finally : <EOL> hdf5file . close ( ) <EOL> self . assertGreaterEqual ( len ( ncomments ) , <NUM_LIT:1> ) <EOL> self . assertTrue ( any ( x > <NUM_LIT:1> for x in ncomments . values ( ) ) ) <EOL> def my_run_func ( idx ) : <EOL> return '<STR_LIT>' % idx <EOL> def my_set_func ( idx ) : <EOL> return '<STR_LIT>' % idx <EOL> class TestOtherHDF5Settings ( EnvironmentTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> def set_mode ( self ) : <EOL> EnvironmentTest . set_mode ( self ) <EOL> self . mode = '<STR_LIT>' <EOL> self . multiproc = False <EOL> self . ncores = <NUM_LIT:1> <EOL> self . use_pool = True <EOL> self . pandas_format = '<STR_LIT>' <EOL> self . pandas_append = True <EOL> self . complib = '<STR_LIT>' <EOL> self . complevel = <NUM_LIT:2> <EOL> self . shuffle = False <EOL> self . fletcher32 = False <EOL> self . encoding = '<STR_LIT>' <EOL> self . graceful_exit = True <EOL> class TestOtherHDF5Settings2 ( EnvironmentTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> def set_mode ( self ) : <EOL> EnvironmentTest . set_mode ( self ) <EOL> self . mode = '<STR_LIT>' <EOL> self . multiproc = False <EOL> self . ncores = <NUM_LIT:1> <EOL> self . use_pool = True <EOL> self . pandas_format = '<STR_LIT>' <EOL> self . pandas_append = False <EOL> self . complib = '<STR_LIT>' <EOL> self . complevel = <NUM_LIT:2> <EOL> self . shuffle = False <EOL> self . fletcher32 = True <EOL> self . encoding = '<STR_LIT>' <EOL> self . wildcard_functions = { ( '<STR_LIT:$>' , '<STR_LIT>' ) : my_run_func , ( '<STR_LIT>' , '<STR_LIT>' ) : my_set_func } <EOL> class ResultSortTest ( TrajectoryComparator ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> def set_mode ( self ) : <EOL> self . mode = '<STR_LIT>' <EOL> self . multiproc = False <EOL> self . ncores = <NUM_LIT:1> <EOL> self . use_pool = True <EOL> self . log_stdout = False <EOL> self . freeze_input = False <EOL> self . use_scoop = False <EOL> self . log_config = True <EOL> self . port = None <EOL> self . graceful_exit = True <EOL> def tearDown ( self ) : <EOL> self . env . f_disable_logging ( ) <EOL> super ( ResultSortTest , self ) . tearDown ( ) <EOL> def setUp ( self ) : <EOL> self . set_mode ( ) <EOL> self . filename = make_temp_dir ( os . path . join ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . trajname = make_trajectory_name ( self ) <EOL> env = Environment ( trajectory = self . trajname , filename = self . filename , <EOL> file_title = self . trajname , <EOL> log_stdout = self . log_stdout , <EOL> log_config = get_log_config ( ) if self . log_config else None , <EOL> multiproc = self . multiproc , <EOL> wrap_mode = self . mode , <EOL> ncores = self . ncores , <EOL> use_pool = self . use_pool , <EOL> use_scoop = self . use_scoop , <EOL> port = self . port , <EOL> freeze_input = self . freeze_input , <EOL> graceful_exit = self . graceful_exit ) <EOL> traj = env . v_trajectory <EOL> traj . v_standard_parameter = Parameter <EOL> traj . f_add_parameter ( '<STR_LIT:x>' , <NUM_LIT> ) <EOL> traj . f_add_parameter ( '<STR_LIT:y>' , <NUM_LIT> ) <EOL> self . env = env <EOL> self . traj = traj <EOL> def load_trajectory ( self , trajectory_index = None , trajectory_name = None , as_new = False , how = <NUM_LIT:2> ) : <EOL> newtraj = Trajectory ( ) <EOL> newtraj . v_storage_service = HDF5StorageService ( filename = self . filename ) <EOL> newtraj . f_load ( name = trajectory_name , index = trajectory_index , as_new = as_new , <EOL> load_derived_parameters = how , load_results = how ) <EOL> return newtraj <EOL> def explore ( self , traj ) : <EOL> self . explore_dict = { '<STR_LIT:x>' : [ - <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> ] , '<STR_LIT:y>' : [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:3> ] } <EOL> traj . f_explore ( self . explore_dict ) <EOL> def explore_cartesian ( self , traj ) : <EOL> self . explore_dict = cartesian_product ( { '<STR_LIT:x>' : [ - <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] , '<STR_LIT:y>' : [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:4> ] } ) <EOL> traj . f_explore ( self . explore_dict ) <EOL> def expand ( self , traj ) : <EOL> self . expand_dict = { '<STR_LIT:x>' : [ <NUM_LIT:10> , <NUM_LIT:11> , <NUM_LIT:12> , <NUM_LIT> ] , '<STR_LIT:y>' : [ <NUM_LIT:11> , <NUM_LIT:11> , <NUM_LIT:12> , <NUM_LIT:12> , <NUM_LIT> ] } <EOL> with self . assertRaises ( ValueError ) : <EOL> traj . f_expand ( self . expand_dict ) <EOL> self . expand_dict = { '<STR_LIT:x>' : [ <NUM_LIT:10> , <NUM_LIT:11> , <NUM_LIT:12> , <NUM_LIT> ] , '<STR_LIT:y>' : [ <NUM_LIT:11> , <NUM_LIT:11> , <NUM_LIT:12> , <NUM_LIT:12> ] } <EOL> traj . f_expand ( self . expand_dict ) <EOL> def test_if_results_are_sorted_correctly_manual_runs ( self ) : <EOL> self . explore ( self . traj ) <EOL> self . traj . f_store ( only_init = True ) <EOL> man_multiply = manual_run ( ) ( multiply_with_storing ) <EOL> for idx in self . traj . f_iter_runs ( yields = '<STR_LIT>' ) : <EOL> self . assertTrue ( isinstance ( idx , int ) ) <EOL> man_multiply ( self . traj ) <EOL> traj = self . traj <EOL> traj . f_store ( ) <EOL> self . assertTrue ( len ( traj ) , <NUM_LIT:5> ) <EOL> self . assertTrue ( len ( traj ) == len ( compat . listvalues ( self . explore_dict ) [ <NUM_LIT:0> ] ) ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . check_if_z_is_correct ( traj ) <EOL> newtraj = self . load_trajectory ( trajectory_name = self . traj . v_name , as_new = False ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . compare_trajectories ( self . traj , newtraj ) <EOL> def test_if_results_are_sorted_correctly_using_map ( self ) : <EOL> self . explore ( self . traj ) <EOL> args1 = [ <NUM_LIT:10> * x for x in range ( len ( self . traj ) ) ] <EOL> args2 = [ <NUM_LIT:100> * x for x in range ( len ( self . traj ) ) ] <EOL> args3 = list ( range ( len ( self . traj ) ) ) <EOL> results = self . env . f_run_map ( multiply_args , args1 , arg2 = args2 , arg3 = args3 ) <EOL> self . assertEqual ( len ( results ) , len ( self . traj ) ) <EOL> traj = self . traj <EOL> self . assertTrue ( len ( traj ) == len ( compat . listvalues ( self . explore_dict ) [ <NUM_LIT:0> ] ) ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . check_if_z_is_correct_map ( traj , args1 , args2 , args3 ) <EOL> for res in results : <EOL> self . assertEqual ( len ( res ) , <NUM_LIT:2> ) <EOL> self . assertTrue ( isinstance ( res [ <NUM_LIT:0> ] , int ) ) <EOL> self . assertTrue ( isinstance ( res [ <NUM_LIT:1> ] , int ) ) <EOL> idx = res [ <NUM_LIT:0> ] <EOL> self . assertEqual ( self . traj . res . runs [ idx ] . z , res [ <NUM_LIT:1> ] ) <EOL> newtraj = self . load_trajectory ( trajectory_name = self . traj . v_name , as_new = False ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . assertEqual ( len ( traj ) , <NUM_LIT:5> ) <EOL> self . compare_trajectories ( self . traj , newtraj ) <EOL> def test_if_results_are_sorted_correctly ( self ) : <EOL> self . explore ( self . traj ) <EOL> results = self . env . f_run ( multiply ) <EOL> self . are_results_in_order ( results ) <EOL> self . assertEqual ( len ( results ) , len ( self . traj ) ) <EOL> traj = self . traj <EOL> self . assertTrue ( len ( traj ) == len ( compat . listvalues ( self . explore_dict ) [ <NUM_LIT:0> ] ) ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . check_if_z_is_correct ( traj ) <EOL> for res in results : <EOL> self . assertEqual ( len ( res ) , <NUM_LIT:2> ) <EOL> self . assertTrue ( isinstance ( res [ <NUM_LIT:0> ] , int ) ) <EOL> self . assertTrue ( isinstance ( res [ <NUM_LIT:1> ] , int ) ) <EOL> idx = res [ <NUM_LIT:0> ] <EOL> self . assertEqual ( self . traj . res . runs [ idx ] . z , res [ <NUM_LIT:1> ] ) <EOL> newtraj = self . load_trajectory ( trajectory_name = self . traj . v_name , as_new = False ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . compare_trajectories ( self . traj , newtraj ) <EOL> def test_graceful_exit ( self ) : <EOL> self . explore_cartesian ( self . traj ) <EOL> results = self . env . f_run ( multiply_with_graceful_exit ) <EOL> self . are_results_in_order ( results ) <EOL> self . assertFalse ( self . traj . f_is_completed ( ) ) <EOL> def test_f_iter_runs ( self ) : <EOL> self . explore ( self . traj ) <EOL> results = self . env . f_run ( multiply ) <EOL> self . are_results_in_order ( results ) <EOL> traj = self . traj <EOL> self . assertTrue ( len ( traj ) == len ( compat . listvalues ( self . explore_dict ) [ <NUM_LIT:0> ] ) ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . check_if_z_is_correct ( traj ) <EOL> newtraj = self . load_trajectory ( trajectory_name = self . traj . v_name , as_new = False ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> for idx , run_name in enumerate ( self . traj . f_iter_runs ( ) ) : <EOL> newtraj . v_crun = run_name <EOL> self . traj . v_idx = idx <EOL> newtraj . v_idx = idx <EOL> nameset = set ( ( x . v_name for x in traj . f_iter_nodes ( predicate = ( idx , ) ) ) ) <EOL> self . assertTrue ( '<STR_LIT>' % ( idx + <NUM_LIT:1> ) not in nameset ) <EOL> self . assertTrue ( '<STR_LIT>' % idx in nameset ) <EOL> self . assertTrue ( traj . v_crun == run_name ) <EOL> self . assertTrue ( newtraj . crun . z == traj . x * traj . y , '<STR_LIT>' % <EOL> ( str ( newtraj . crun . z ) , str ( traj . x ) , str ( traj . y ) ) ) <EOL> for idx , traj in enumerate ( self . traj . f_iter_runs ( yields = '<STR_LIT>' ) ) : <EOL> run_name = traj . f_idx_to_run ( idx ) <EOL> self . assertTrue ( traj is self . traj ) <EOL> newtraj . v_crun = run_name <EOL> self . traj . v_idx = idx <EOL> newtraj . v_idx = idx <EOL> nameset = set ( ( x . v_name for x in traj . f_iter_nodes ( predicate = ( idx , ) ) ) ) <EOL> self . assertTrue ( '<STR_LIT>' % ( idx + <NUM_LIT:1> ) not in nameset ) <EOL> self . assertTrue ( '<STR_LIT>' % idx in nameset ) <EOL> self . assertTrue ( traj . v_crun == run_name ) <EOL> self . assertTrue ( newtraj . crun . z == traj . x * traj . y , '<STR_LIT>' % <EOL> ( str ( newtraj . crun . z ) , str ( traj . x ) , str ( traj . y ) ) ) <EOL> for idx , traj in enumerate ( self . traj . f_iter_runs ( yields = '<STR_LIT>' ) ) : <EOL> run_name = traj . f_idx_to_run ( idx ) <EOL> self . assertTrue ( traj is not self . traj ) <EOL> newtraj . v_crun = run_name <EOL> self . traj . v_idx = idx <EOL> newtraj . v_idx = idx <EOL> nameset = set ( ( x . v_name for x in traj . f_iter_nodes ( predicate = ( idx , ) ) ) ) <EOL> self . assertTrue ( '<STR_LIT>' % ( idx + <NUM_LIT:1> ) not in nameset ) <EOL> self . assertTrue ( '<STR_LIT>' % idx in nameset ) <EOL> self . assertTrue ( traj . v_crun == run_name ) <EOL> self . assertTrue ( newtraj . crun . z == traj . x * traj . y , '<STR_LIT>' % <EOL> ( str ( newtraj . crun . z ) , str ( traj . x ) , str ( traj . y ) ) ) <EOL> traj = self . traj <EOL> self . assertTrue ( traj . v_idx == - <NUM_LIT:1> ) <EOL> self . assertTrue ( traj . v_crun is None ) <EOL> self . assertTrue ( traj . v_crun_ == pypetconstants . RUN_NAME_DUMMY ) <EOL> self . assertTrue ( newtraj . v_idx == idx ) <EOL> def test_f_iter_runs_auto_load ( self ) : <EOL> self . explore ( self . traj ) <EOL> results = self . env . f_run ( multiply ) <EOL> self . are_results_in_order ( results ) <EOL> traj = self . traj <EOL> self . assertTrue ( len ( traj ) == len ( compat . listvalues ( self . explore_dict ) [ <NUM_LIT:0> ] ) ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . check_if_z_is_correct ( traj ) <EOL> newtraj = Trajectory ( ) <EOL> newtraj . v_storage_service = HDF5StorageService ( filename = self . filename ) <EOL> newtraj . f_load ( name = self . traj . v_name , index = None , as_new = False , load_data = <NUM_LIT:0> ) <EOL> newtraj . v_auto_load = True <EOL> newtraj . par . f_load_child ( '<STR_LIT:y>' , load_data = <NUM_LIT:1> ) <EOL> for idx , run_name in enumerate ( self . traj . f_iter_runs ( ) ) : <EOL> newtraj . v_crun = run_name <EOL> self . traj . v_idx = idx <EOL> newtraj . v_idx = idx <EOL> nameset = set ( ( x . v_name for x in traj . f_iter_nodes ( predicate = ( idx , ) ) ) ) <EOL> self . assertTrue ( '<STR_LIT>' % ( idx + <NUM_LIT:1> ) not in nameset ) <EOL> self . assertTrue ( '<STR_LIT>' % idx in nameset ) <EOL> self . assertTrue ( traj . v_crun == run_name ) <EOL> self . assertTrue ( newtraj . res . runs . crun . z == newtraj . par . x * newtraj . par . y , '<STR_LIT>' % <EOL> ( str ( newtraj . crun . z ) , str ( newtraj . x ) , str ( newtraj . y ) ) ) <EOL> traj = self . traj <EOL> self . assertTrue ( traj . v_idx == - <NUM_LIT:1> ) <EOL> self . assertTrue ( traj . v_crun is None ) <EOL> self . assertTrue ( traj . v_crun_ == pypetconstants . RUN_NAME_DUMMY ) <EOL> self . assertTrue ( newtraj . v_idx == idx ) <EOL> def test_expand ( self ) : <EOL> self . explore ( self . traj ) <EOL> results = self . env . f_run ( multiply ) <EOL> self . are_results_in_order ( results ) <EOL> get_root_logger ( ) . info ( results ) <EOL> traj = self . traj <EOL> self . assertEqual ( len ( traj ) , len ( list ( compat . listvalues ( self . explore_dict ) [ <NUM_LIT:0> ] ) ) ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . check_if_z_is_correct ( traj ) <EOL> traj_name = self . env . v_trajectory . v_name <EOL> del self . env <EOL> self . env = Environment ( trajectory = self . traj , <EOL> log_stdout = False , <EOL> log_config = get_log_config ( ) ) <EOL> self . traj = self . env . v_trajectory <EOL> self . traj . f_load ( name = traj_name ) <EOL> self . expand ( self . traj ) <EOL> results = self . env . f_run ( multiply ) <EOL> self . are_results_in_order ( results ) <EOL> traj = self . traj <EOL> self . assertTrue ( len ( traj ) == len ( compat . listvalues ( self . expand_dict ) [ <NUM_LIT:0> ] ) + len ( compat . listvalues ( self . explore_dict ) [ <NUM_LIT:0> ] ) ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . check_if_z_is_correct ( traj ) <EOL> newtraj = self . load_trajectory ( trajectory_name = self . traj . v_name , as_new = False ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . compare_trajectories ( self . traj , newtraj ) <EOL> def test_expand_after_reload ( self ) : <EOL> self . explore ( self . traj ) <EOL> results = self . env . f_run ( multiply ) <EOL> self . are_results_in_order ( results ) <EOL> traj = self . traj <EOL> self . assertTrue ( len ( traj ) == len ( compat . listvalues ( self . explore_dict ) [ <NUM_LIT:0> ] ) ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . check_if_z_is_correct ( traj ) <EOL> self . expand ( self . traj ) <EOL> self . env . f_run ( multiply ) <EOL> traj = self . traj <EOL> self . assertTrue ( len ( traj ) == len ( compat . listvalues ( self . expand_dict ) [ <NUM_LIT:0> ] ) + len ( compat . listvalues ( self . explore_dict ) [ <NUM_LIT:0> ] ) ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . check_if_z_is_correct ( traj ) <EOL> newtraj = self . load_trajectory ( trajectory_name = self . traj . v_name , as_new = False ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . compare_trajectories ( self . traj , newtraj ) <EOL> def check_if_z_is_correct_map ( self , traj , args1 , args2 , args3 ) : <EOL> for x , arg1 , arg2 , arg3 in zip ( range ( len ( traj ) ) , args1 , args2 , args3 ) : <EOL> traj . v_idx = x <EOL> self . assertTrue ( traj . crun . z == traj . x * traj . y + arg1 + arg2 + arg3 , '<STR_LIT>' % <EOL> ( str ( traj . crun . z ) , str ( traj . x ) , str ( traj . y ) ) ) <EOL> traj . v_idx = - <NUM_LIT:1> <EOL> def check_if_z_is_correct ( self , traj ) : <EOL> traj . v_shortcuts = False <EOL> for x in range ( len ( traj ) ) : <EOL> traj . v_idx = x <EOL> z = traj . res . runs . crun . z <EOL> x = traj . par . x <EOL> y = traj . par . y <EOL> self . assertTrue ( z == x * y , '<STR_LIT>' % <EOL> ( str ( z ) , str ( x ) , str ( y ) ) ) <EOL> traj . v_idx = - <NUM_LIT:1> <EOL> traj . v_shortcuts = True <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> opt_args = parse_args ( ) <EOL> run_suite ( ** opt_args ) </s>
