# Path: modred/util.py
# class UndefinedError(Exception): pass
# class InnerProductBlock(object):
# def atleast_2d_row(array):
# def atleast_2d_col(array):
# def make_iterable(arg):
# def flatten_list(my_list):
# def save_array_text(array, file_name, delimiter=None):
# def load_array_text(file_name, delimiter=None, is_complex=False):
# def get_file_list(directory, file_extension=None):
# def get_data_members(obj):
# def sum_arrays(arr1, arr2):
# def sum_lists(list1, list2):
# def smart_eq(arg1, arg2):
#     def __init__(self, inner_product):
#     def __call__(self, vecs1, vecs2):
# def svd(array, atol=1e-13, rtol=None):
# def eigh(array, atol=1e-13, rtol=None, is_positive_definite=False):
# def eig_biorthog(array, scale_choice='left'):
# def balanced_truncation(
#     A, B, C, order=None, return_sing_vals=False):
# def drss(num_states, num_inputs, num_outputs):
# def rss(num_states, num_inputs, num_outputs):
# def lsim(A, B, C, inputs, initial_condition=None):
# def impulse(A, B, C, num_time_steps=None):
# def load_signals(signal_path, delimiter=None):
# def load_multiple_signals(signal_paths, delimiter=None):
# def Hankel(first_col, last_row=None):
# def Hankel_chunks(first_col_chunks, last_row_chunks=None):
#     V = V_conj_T.conj().T
#     U = U[:, :num_nonzeros]
#     V = V[:, :num_nonzeros]
#     S = S[:num_nonzeros]
#     A = np.array(A)
#     B = np.array(B)
#     C = np.array(C)
#     U, E, V = svd(Lo.transpose().dot(Lc))
#     SL = Lo.dot(U[:, :order]).dot(np.diag(E[:order]**-0.5))
#     SR = Lc.dot(V[:, :order]).dot(np.diag(E[:order]**-0.5))
#     A = np.real(
#         np.linalg.inv(eig_vecs).dot(np.diag(eig_vals).dot(eig_vecs)))
#     B = np.random.normal(0, 1., (num_states, num_inputs))
#     C = np.random.normal(0, 1., (num_outputs, num_states))
#     A = np.linalg.inv(transformation).dot(np.diag(e_vals)).dot(transformation)
#     B = np.random.random((num_states, num_inputs))
#     C = np.random.random((num_outputs, num_states))
#     A = np.array(A)
#     B = np.array(B)
#     C = np.array(C)
# 
# Path: modred/parallel.py
# def get_hostname():
# def get_node_ID():
# def get_num_nodes():
# def get_num_MPI_workers():
# def get_rank():
# def get_num_procs():
# def is_distributed():
# def is_rank_zero():
# def barrier():
# def print_from_rank_zero(msg, output_channel='stdout'):
# def call_from_rank_zero(func, *args, **kwargs):
# def bcast(vals):
# def call_and_bcast(func, *args, **kwargs):
# def find_assignments(tasks, task_weights=None):
# def check_for_empty_tasks(task_assignments):
# 
# Path: modred/py2to3.py
# def run_script(path, globals=None, locals=None):
# def print_stdout(msg):
# def print_stderr(msg):
# def print_msg(msg, output_channel='stdout'):


import unittest
import os
import numpy as np
from shutil import rmtree   # For deleting directories and their contents
from os.path import join
from modred import util, parallel
from modred.py2to3 import range
        # Check that error is raised for invalid scale choice
        self.assertRaises(
            ValueError, util.eig_biorthog, array, **{'scale_choice':'invalid'})


    #@unittest.skip('Testing something else.')
    @unittest.skipIf(parallel.is_distributed(), 'Only load data in serial')
    def test_load_impulse_outputs(self):
        """
        Test loading multiple signal files in [t sig1 sig2 ...] format.

        Creates signals, saves them, loads them, tests are equal to the
        originals.
        """
        signal_path = join(self.test_dir, 'file%03d.txt')
        for num_paths in [1, 4]:
            for num_signals in [1, 2, 4, 5]:
                for num_time_steps in [1, 10, 100]:
                    all_signals_true = np.random.random((num_paths,
                        num_time_steps, num_signals))
                    # Time steps need not be sequential
                    time_values_true = np.random.random(num_time_steps)

                    signal_paths = []
                    # Save signals to file
                    for path_num in range(num_paths):
                        signal_paths.append(signal_path%path_num)
                        data_to_save = np.concatenate( \
                          (time_values_true.reshape(len(time_values_true), 1),
                          all_signals_true[path_num]), axis=1)

# Path: modred/util.py
# class UndefinedError(Exception): pass
# class InnerProductBlock(object):
# def atleast_2d_row(array):
# def atleast_2d_col(array):
# def make_iterable(arg):
# def flatten_list(my_list):
# def save_array_text(array, file_name, delimiter=None):
# def load_array_text(file_name, delimiter=None, is_complex=False):
# def get_file_list(directory, file_extension=None):
# def get_data_members(obj):
# def sum_arrays(arr1, arr2):
# def sum_lists(list1, list2):
# def smart_eq(arg1, arg2):
#     def __init__(self, inner_product):
#     def __call__(self, vecs1, vecs2):
# def svd(array, atol=1e-13, rtol=None):
# def eigh(array, atol=1e-13, rtol=None, is_positive_definite=False):
# def eig_biorthog(array, scale_choice='left'):
# def balanced_truncation(
#     A, B, C, order=None, return_sing_vals=False):
# def drss(num_states, num_inputs, num_outputs):
# def rss(num_states, num_inputs, num_outputs):
# def lsim(A, B, C, inputs, initial_condition=None):
# def impulse(A, B, C, num_time_steps=None):
# def load_signals(signal_path, delimiter=None):
# def load_multiple_signals(signal_paths, delimiter=None):
# def Hankel(first_col, last_row=None):
# def Hankel_chunks(first_col_chunks, last_row_chunks=None):
#     V = V_conj_T.conj().T
#     U = U[:, :num_nonzeros]
#     V = V[:, :num_nonzeros]
#     S = S[:num_nonzeros]
#     A = np.array(A)
#     B = np.array(B)
#     C = np.array(C)
#     U, E, V = svd(Lo.transpose().dot(Lc))
#     SL = Lo.dot(U[:, :order]).dot(np.diag(E[:order]**-0.5))
#     SR = Lc.dot(V[:, :order]).dot(np.diag(E[:order]**-0.5))
#     A = np.real(
#         np.linalg.inv(eig_vecs).dot(np.diag(eig_vals).dot(eig_vecs)))
#     B = np.random.normal(0, 1., (num_states, num_inputs))
#     C = np.random.normal(0, 1., (num_outputs, num_states))
#     A = np.linalg.inv(transformation).dot(np.diag(e_vals)).dot(transformation)
#     B = np.random.random((num_states, num_inputs))
#     C = np.random.random((num_outputs, num_states))
#     A = np.array(A)
#     B = np.array(B)
#     C = np.array(C)
# 
# Path: modred/parallel.py
# def get_hostname():
# def get_node_ID():
# def get_num_nodes():
# def get_num_MPI_workers():
# def get_rank():
# def get_num_procs():
# def is_distributed():
# def is_rank_zero():
# def barrier():
# def print_from_rank_zero(msg, output_channel='stdout'):
# def call_from_rank_zero(func, *args, **kwargs):
# def bcast(vals):
# def call_and_bcast(func, *args, **kwargs):
# def find_assignments(tasks, task_weights=None):
# def check_for_empty_tasks(task_assignments):
# 
# Path: modred/py2to3.py
# def run_script(path, globals=None, locals=None):
# def print_stdout(msg):
# def print_stderr(msg):
# def print_msg(msg, output_channel='stdout'):


# Path: modred/tests/testutil.py
import unittest
import os
import numpy as np
from shutil import rmtree   # For deleting directories and their contents
from os.path import join
from modred import util, parallel
from modred.py2to3 import range

        # Check that error is raised for invalid scale choice
        self.assertRaises(
            ValueError, util.eig_biorthog, array, **{'scale_choice':'invalid'})


    #@unittest.skip('Testing something else.')
    @unittest.skipIf(parallel.is_distributed(), 'Only load data in serial')
    def test_load_impulse_outputs(self):
        """
        Test loading multiple signal files in [t sig1 sig2 ...] format.

        Creates signals, saves them, loads them, tests are equal to the
        originals.
        """
        signal_path = join(self.test_dir, 'file%03d.txt')
        for num_paths in [1, 4]:
            for num_signals in [1, 2, 4, 5]:
                for num_time_steps in [1, 10, 100]:
                    all_signals_true = np.random.random((num_paths,
                        num_time_steps, num_signals))
                    # Time steps need not be sequential
                    time_values_true = np.random.random(num_time_steps)

                    signal_paths = []
                    # Save signals to file
                    for path_num in range(num_paths):
                        signal_paths.append(signal_path%path_num)
                        data_to_save = np.concatenate( \
                          (time_values_true.reshape(len(time_values_true), 1),
                          all_signals_true[path_num]), axis=1)

                        util.save_array_text(data_to_save, signal_path%path_num)