<commit_before>#!/usr/bin/env python
"""Test POD module"""
from __future__ import division
from future.builtins import range

import unittest
import os
import numpy as np
from os.path import join
from shutil import rmtree
import copy


import modred.parallel as parallel_mod
_parallel = parallel_mod.parallel_default_instance

from modred.pod import *
from modred.vectorspace import *
import modred.vectors as V
from modred import util



@unittest.skipIf(_parallel.is_distributed(), 'Serial only.')
class TestPODArraysFunctions(unittest.TestCase):
    def setUp(self):
        self.mode_indices = [2, 4, 6, 3]
        self.num_vecs = 10
        self.num_states = 30

    def test_compute_modes(self):
        ws = np.identity(self.num_states)
        tol = 1e-6
        weights_full = np.mat(np.random.random((self.num_states, self.num_states)))
        weights_full = np.triu(weights_full) + np.triu(weights_full, 1).conj().T
        weights_full = weights_full*weights_full
        weights_diag = np.random.random(self.num_states)
        weights_list = [None, weights_diag, weights_full]
        vec_array = np.random.random((self.num_states, self.num_vecs))
        for weights in weights_list:
            IP = VectorSpaceMatrices(weights=weights).compute_inner_product_mat
            correlation_mat_true = IP(vec_array, vec_array)
            eigvals_true, eigvecs_true = util.eigh(correlation_mat_true)
            build_coeff_mat_true = eigvecs_true * np.mat(np.diag(
                eigvals_true**-0.5))
            modes_true = vec_array.dot(build_coeff_mat_true)
            
            modes, eigvals, eigvecs, correlation_mat = \
                compute_POD_matrices_snaps_method(vec_array, self.mode_indices, 
                inner_product_weights=weights, return_all=True)
            
            np.testing.assert_allclose(eigvals, eigvals_true, rtol=tol)
            np.testing.assert_allclose(eigvecs, eigvecs_true)
            np.testing.assert_allclose(correlation_mat, correlation_mat_true)
            np.testing.assert_allclose(modes, modes_true[:,self.mode_indices])
                        
            modes, eigvals, eigvecs = \
                compute_POD_matrices_direct_method(vec_array, self.mode_indices, 
                inner_product_weights=weights, return_all=True)
            
            np.testing.assert_allclose(eigvals, eigvals_true)
            np.testing.assert_allclose(np.abs(eigvecs), np.abs(eigvecs_true))
            np.testing.assert_allclose(np.abs(modes), np.abs(modes_true[:,self.mode_indices]))
            
            


class TestPODHandles(unittest.TestCase):
    def setUp(self):
        self.test_dir = 'DELETE_ME_test_files_pod'
        if not os.access('.', os.W_OK):
            raise RuntimeError('Cannot write to current directory')
        if not os.path.isdir(self.test_dir) and _parallel.is_rank_zero():        
            os.mkdir(self.test_dir)
        self.mode_indices = [2, 4, 3, 6]
        self.num_vecs = 10
        self.num_states = 30
        self.vec_array = _parallel.call_and_bcast(np.random.random, 
            (self.num_states, self.num_vecs))
        self.correlation_mat_true = self.vec_array.conj().transpose().dot(
            self.vec_array)
    
        self.eigvals_true, self.eigvecs_true = \
            _parallel.call_and_bcast(util.eigh, self.correlation_mat_true)

        self.mode_array = np.dot(self.vec_array, np.dot(self.eigvecs_true,
            np.diag(self.eigvals_true**-0.5)))
        self.vec_path = join(self.test_dir, 'vec_%03d.txt')
        self.vec_handles = [V.VecHandleArrayText(self.vec_path%i)<commit_msg>Fixed POD test.  No longer fails due to error.<commit_after>#!/usr/bin/env python
"""Test POD module"""
from __future__ import division
from future.builtins import range

import unittest
import os
import numpy as np
from os.path import join
from shutil import rmtree
import copy


import modred.parallel as parallel_mod
_parallel = parallel_mod.parallel_default_instance

from modred.pod import *
from modred.vectorspace import *
import modred.vectors as V
from modred import util



@unittest.skipIf(_parallel.is_distributed(), 'Serial only.')
class TestPODArraysFunctions(unittest.TestCase):
    def setUp(self):
        self.mode_indices = [2, 4, 6, 3]
        self.num_vecs = 10
        self.num_states = 30

    def test_compute_modes(self):
        ws = np.identity(self.num_states)
        tol = 1e-6

        # Generate different inner product weights.  Each weight matrix should
        # be positive definite semidefinite.
        weights_full = np.mat(
            np.random.random((self.num_states, self.num_states)))
        weights_full = 0.5 * (weights_full + weights_full.T)
        weights_full = weights_full + self.num_states * np.eye(self.num_states)
        weights_diag = np.random.random(self.num_states)
        weights_list = [None, weights_diag, weights_full]
        vec_array = np.random.random((self.num_states, self.num_vecs))
        for weights in weights_list:
            IP = VectorSpaceMatrices(weights=weights).compute_inner_product_mat
            correlation_mat_true = IP(vec_array, vec_array)
            eigvals_true, eigvecs_true = util.eigh(correlation_mat_true)
            build_coeff_mat_true = eigvecs_true * np.mat(np.diag(
                eigvals_true**-0.5))
            modes_true = vec_array.dot(build_coeff_mat_true)
            
            modes, eigvals, eigvecs, correlation_mat = \
                compute_POD_matrices_snaps_method(vec_array, self.mode_indices, 
                inner_product_weights=weights, return_all=True)
            
            np.testing.assert_allclose(eigvals, eigvals_true, rtol=tol)
            np.testing.assert_allclose(eigvecs, eigvecs_true)
            np.testing.assert_allclose(correlation_mat, correlation_mat_true)
            np.testing.assert_allclose(modes, modes_true[:,self.mode_indices])
                        
            modes, eigvals, eigvecs = \
                compute_POD_matrices_direct_method(
                vec_array, self.mode_indices, inner_product_weights=weights, 
                return_all=True)
            
            np.testing.assert_allclose(eigvals, eigvals_true)
            np.testing.assert_allclose(np.abs(eigvecs), np.abs(eigvecs_true))
            np.testing.assert_allclose(
                np.abs(modes), np.abs(modes_true[:,self.mode_indices]))
            
            


class TestPODHandles(unittest.TestCase):
    def setUp(self):
        self.test_dir = 'DELETE_ME_test_files_pod'
        if not os.access('.', os.W_OK):
            raise RuntimeError('Cannot write to current directory')
        if not os.path.isdir(self.test_dir) and _parallel.is_rank_zero():        
            os.mkdir(self.test_dir)
        self.mode_indices = [2, 4, 3, 6]
        self.num_vecs = 10
        self.num_states = 30
        self.vec_array = _parallel.call_and_bcast(np.random.random, 
            (self.num_states, self.num_vecs))
        self.correlation_mat_true = self.vec_array.conj().transpose().dot(
            self.vec_array)
    
        self.eigvals_true, self.eigvecs_true = \
            _parallel.call_and_bcast(util.eigh, self.correlation_mat_true)

        self.mode_array = np.dot(self.vec_array, np.dot(self.eigvecs_true,
            np.diag(self.eigvals_true**-0.5)))
        self.vec_path = join(self.test_dir, 'vec_%03d.txt')
        self.vec_handles = [V.VecHandleArrayText(self.vec_path%i)
{'fraction_non_alphanumeric': 0.09453668987680772, 'fraction_numerical': 0.006829137653990359, 'mean_word_length': 2.625728155339806, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '776971', 'n_tokens_mistral': 2472, 'n_tokens_neox': 2377, 'n_words': 405}