<commit_before>'''
Created on 23.01.2015

@author: marscher
'''
import mdtraj
import os
import tempfile
import unittest

from pyemma.coordinates.io.feature_reader import FeatureReader
from pyemma.util.log import getLogger
import pkg_resources

import numpy as np
from pyemma.coordinates.api import feature_reader, discretizer, tica

log = getLogger('TestFeatureReader')


class TestFeatureReader(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        c = super(TestFeatureReader, cls).setUpClass()
        # create a fake trajectory which has 3 atoms and coordinates are just a range
        # over all frames.
        cls.trajfile = tempfile.mktemp('.xtc')
        cls.n_frames = 1000
        cls.xyz = np.random.random(cls.n_frames * 3 * 3).reshape((cls.n_frames, 3, 3))
        log.debug("shape traj: %s" % str(cls.xyz.shape))
        cls.topfile = pkg_resources.resource_filename(
            'pyemma.coordinates.tests.test_featurereader', 'data/test.pdb')
        t = mdtraj.load(cls.topfile)
        t.xyz = cls.xyz
        t.time = np.arange(cls.n_frames)
        t.save(cls.trajfile)
        return c

    @classmethod
    def tearDownClass(cls):
        os.unlink(cls.trajfile)
        super(TestFeatureReader, cls).tearDownClass()

    def testIteratorAccess(self):
        reader = FeatureReader(self.trajfile, self.topfile)

        frames = 0
        data = []
        for i, X in reader:
            frames += X.shape[0]
            data.append(X)

        # restore shape of input
        data = np.array(data).reshape(self.xyz.shape)

        self.assertEqual(frames, reader.trajectory_lengths()[0])
        self.assertTrue(np.allclose(data, self.xyz))

    def testIteratorAccess2(self):
        reader = FeatureReader([self.trajfile, self.trajfile], self.topfile)
        reader.chunksize = 100

        frames = 0
        data = []
        for i, X in reader:
            frames += X.shape[0]
            data.append(X)
        self.assertEqual(frames, reader.trajectory_lengths()[0] * 2)
        # restore shape of input
        data = np.array(
            data[0:reader.trajectory_lengths()[0] / reader.chunksize]).reshape(self.xyz.shape)

        self.assertTrue(np.allclose(data, self.xyz))

    def testTimeLaggedIterator(self):
        lag = 10
        reader = FeatureReader(self.trajfile, self.topfile)
        reader.lag = lag
        frames = 0
        data = []
        lagged = []
        for _, X, Y in reader:
            frames += X.shape[0]
            data.append(X)
            lagged.append(Y)

        assert len(data) == len(lagged)
        merged_lagged = np.concatenate(lagged,axis=0)  # .reshape(self.xyz.shape)

        # reproduce outcome
        xyz_s = self.xyz.shape
        fake_lagged = np.empty((xyz_s[0]-lag,xyz_s[1],xyz_s[2]))
        fake_lagged = self.xyz[lag:]

        self.assertTrue(np.allclose(merged_lagged, fake_lagged))

        # restore shape of input
        data = np.array(data).reshape(self.xyz.shape)

        self.assertEqual(frames, reader.trajectory_lengths()[0])
        self.assertTrue(np.allclose(data, self.xyz))
        
    def test_with_pipeline_time_lagged(self):
        reader= feature_reader(self.trajfile, self.topfile)
        reader.featurizer.distances([[0,1], [0,2]])
        t = tica(dim=2,lag=1)
        d = discretizer(reader, t)
        d.run()

    <EMAIL>("")
    def testTimeLaggedAccess(self):
        # each frame has 2 atoms with 3 coords = 6 coords per frame.
        # coords are sequential through all frames and start with 0.

        lags = [2, 200]

        chunksizes = [1, 100]

        for lag in lags:
            for chunksize in chunksizes:
                log.info("chunksize=%i\tlag=%i" % (chunksize, lag))

                lagged_chunks = []
                reader = FeatureReader(self.trajfile, self.topfile)
                reader.chunksize = chunksize
                reader.lag = lag
                for _, _, y in reader:
                    lagged_chunks.append(y)

                coords = self.xyz

                for ii, c in enumerate(lagged_chunks[:-1]):
                    # all despite last chunk shall have chunksize
                    self.assertTrue(c.shape[0] <= chunksize)
                    # first lagged chunk should start at lag and stop at chunksize +
                    # lag
                    ind1 = ii * chunksize + lag<commit_msg>[tests] fixed bug in test_featurereader<commit_after>'''
Created on 23.01.2015

@author: marscher
'''
import mdtraj
import os
import tempfile
import unittest

from pyemma.coordinates.io.feature_reader import FeatureReader
from pyemma.util.log import getLogger
import pkg_resources

import numpy as np
from pyemma.coordinates.api import feature_reader, discretizer, tica

log = getLogger('TestFeatureReader')


class TestFeatureReader(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        c = super(TestFeatureReader, cls).setUpClass()
        # create a fake trajectory which has 3 atoms and coordinates are just a range
        # over all frames.
        cls.trajfile = tempfile.mktemp('.xtc')
        cls.n_frames = 1000
        cls.xyz = np.random.random(cls.n_frames * 3 * 3).reshape((cls.n_frames, 3, 3))
        log.debug("shape traj: %s" % str(cls.xyz.shape))
        cls.topfile = pkg_resources.resource_filename(
            'pyemma.coordinates.tests.test_featurereader', 'data/test.pdb')
        t = mdtraj.load(cls.topfile)
        t.xyz = cls.xyz
        t.time = np.arange(cls.n_frames)
        t.save(cls.trajfile)
        return c

    @classmethod
    def tearDownClass(cls):
        os.unlink(cls.trajfile)
        super(TestFeatureReader, cls).tearDownClass()

    def testIteratorAccess(self):
        reader = FeatureReader(self.trajfile, self.topfile)

        frames = 0
        data = []
        for i, X in reader:
            frames += X.shape[0]
            data.append(X)

        # restore shape of input
        data = np.array(data).reshape(self.xyz.shape)

        self.assertEqual(frames, reader.trajectory_lengths()[0])
        self.assertTrue(np.allclose(data, self.xyz))

    def testIteratorAccess2(self):
        reader = FeatureReader([self.trajfile, self.trajfile], self.topfile)
        reader.chunksize = 100

        frames = 0
        data = []
        for i, X in reader:
            frames += X.shape[0]
            data.append(X)
        self.assertEqual(frames, reader.trajectory_lengths()[0] * 2)
        # restore shape of input
        data = np.array(
            data[0:reader.trajectory_lengths()[0] / reader.chunksize]).reshape(self.xyz.shape)

        self.assertTrue(np.allclose(data, self.xyz))

    def testTimeLaggedIterator(self):
        lag = 10
        reader = FeatureReader(self.trajfile, self.topfile)
        reader.lag = lag
        frames = 0
        data = []
        lagged = []
        for _, X, Y in reader:
            frames += X.shape[0]
            data.append(X)
            lagged.append(Y)

        assert len(data) == len(lagged)
        merged_lagged = np.concatenate(lagged,axis=0)  # .reshape(self.xyz.shape)

        # reproduce outcome
        xyz_s = self.xyz.shape
        fake_lagged = np.empty((xyz_s[0]-lag,xyz_s[1]*xyz_s[2]))
        fake_lagged = self.xyz.reshape((xyz_s[0],-1))[lag:]

        self.assertTrue(np.allclose(merged_lagged, fake_lagged))

        # restore shape of input
        data = np.array(data).reshape(self.xyz.shape)

        self.assertEqual(frames, reader.trajectory_lengths()[0])
        self.assertTrue(np.allclose(data, self.xyz))
        
    def test_with_pipeline_time_lagged(self):
        reader= feature_reader(self.trajfile, self.topfile)
        reader.featurizer.distances([[0,1], [0,2]])
        t = tica(dim=2,lag=1)
        d = discretizer(reader, t)
        d.run()

    <EMAIL>("")
    def testTimeLaggedAccess(self):
        # each frame has 2 atoms with 3 coords = 6 coords per frame.
        # coords are sequential through all frames and start with 0.

        lags = [2, 200]

        chunksizes = [1, 100]

        for lag in lags:
            for chunksize in chunksizes:
                log.info("chunksize=%i\tlag=%i" % (chunksize, lag))

                lagged_chunks = []
                reader = FeatureReader(self.trajfile, self.topfile)
                reader.chunksize = chunksize
                reader.lag = lag
                for _, _, y in reader:
                    lagged_chunks.append(y)

                coords = self.xyz.reshape((self.xyz.shape[0],-1))

                for ii, c in enumerate(lagged_chunks[:-1]):
                    # all despite last chunk shall have chunksize
                    self.assertTrue(c.shape[0] <= chunksize)
                    # first lagged chunk should start at lag and stop at chunksize +
                    # lag
                    ind1 = ii * chunksize + lag
{'fraction_non_alphanumeric': 0.10934505692706573, 'fraction_numerical': 0.013978130988614586, 'mean_word_length': 2.114075114075114, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 5, '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': '5555237', 'n_tokens_mistral': 2902, 'n_tokens_neox': 2724, 'n_words': 603}