<commit_before>#!/usr/bin/env python3

import random
import copy
import numpy as np
import sklearn.pipeline
import json
import unittest
import glob
import warnings
import subprocess
import os
import python.utils as utils
import csv
import h5py


class testClassifieroptions(unittest.TestCase):
    pass


class testHDF5parsing(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        cls.settings_fh = 'test_settings.json'
        cls.settings = utils.get_settings(cls.settings_fh)
        cls.all_subjects = set(['Dog_1',
                             'Dog_2',
                             'Dog_3',
                             'Dog_4',
                             'Dog_5',
                             'Patient_1',
                             'Patient_2'])
        cls.all_types = set(['preictal',
                              'test',
                              'pseudointerictal',
                              'interictal',
                              'pseudopreictal'])

        cls.malformed_feat = 'malformed_feat'
        cls.malformed_file = os.path.join(cls.settings['TRAIN_DATA_PATH'],
                                          "{0}{1}.h5".format(
                                                    cls.malformed_feat,
                                                    cls.settings['VERSION']))

        cls.malformed_feat = h5py.File(cls.malformed_file, 'w')
        cls.malformed_feat.create_dataset('malfie', (10,10))
        cls.malformed_feat.close()

    def test_ioerror_warning(self):
        '''
        Assert a non-existent file correctly raises warning
        '''

        non_existent_feat = 'fake_feat'
        h5_file_name = os.path.join(self.settings['TRAIN_DATA_PATH'],
                                    "{0}{1}.h5".format(non_existent_feat,
                                                       self.settings['VERSION']))

        with warnings.catch_warnings(record=True) as w:
                dummy = utils.parse_matlab_HDF5(non_existent_feat,
                                                self.settings)

                self.assertEqual(len(w), 1, msg="Check there is one and only "
                                                "one warning raised")
                self.assertIs(w[-1].category, UserWarning, msg="Check that "
                                                                "warning raised "
                                                                "is a UserWarning ")
                self.assertEqual(str(w[-1].message),
                                 "{0} does not exist (or is not "
                                 "readable)".format(h5_file_name), msg="Check the "
                                                                        "warning is "
                                                                        "the correct "
                                                                        "format ")

    def test_parse_error_warning(self):
        '''
        Assert malformed HDF5 raises proper warning
        '''


        malformed_feat = 'malformed_feat'
        h5_file_name = os.path.join(self.settings['TRAIN_DATA_PATH'],
                                    "{0}{1}.h5".format(malformed_feat,
                                                       self.settings['VERSION']))

        with warnings.catch_warnings(record=True) as w:

                dummy = utils.parse_matlab_HDF5(malformed_feat,
                                                self.settings)

                self.assertEqual(len(w), 1, msg="Check one and only one error "
                                                "raised")
                self.assertIs(w[-1].category, UserWarning, msg="Check error is "
                                                                "UserWarning")
                self.assertEqual(str(w[-1].message), "Unable to "
                                                     "parse {0}".format(\
                                                                h5_file_name),
                                 msg="Check the warning raised is correct format")


    def test_hdf5_parse(self):
        '''
        Very basic check of parsed object structure containing minimum 3 dicts
        '''
        parsed_HDF5 = utils.parse_matlab_HDF5(self.settings['FEATURES'][0],
                                              self.settings)

        subjects = set(parsed_HDF5.keys())
        self.assertEqual(subjects,
                         self.all_subjects,
                         msg="Check that parsed HDF5 contains all subjects")

        typs = set(parsed_HDF5['Dog_1'].keys())
        self.assertEqual(typs,
                         self.all_types,
                         msg="Check that all ictypes are in parsed file by "
                             "checking the ictypes under Dog_1")

        num_interictal_dog1_segs = len(\
                parsed_HDF5['Dog_1']['interictal'].keys())

        self.assertEqual(num_interictal_dog1_segs,
                         480,
                         msg="Check there is the correct number of segments in "
                             "parsed HDF5s by checking dog1 interictal")

    @classmethod
    def tearDownClass(cls):
        os.unlink(cls.malformed_file)

class testBatchParallel(unittest.TestCase):
    '''
    Class to test the batch parallel script as well as consistency in output
    '''
    @classmethod
    def setUpClass(cls):
        cls.settings_fh_1 = os.path.join('batch_test', 'batch_test1.json')
        cls.settings_fh_2 = os.path.join('batch_test', 'batch_test2.json')
        cls.settings_1 = utils.get_settings(cls.settings_fh_1)
        cls.settings_2 = utils.get_settings(cls.settings_fh_2)

        cls.NULL = open(os.devnull, 'w')
        cls.proc = subprocess.call(['./parallel_batch_run.py',
                                      '-s', 'batch_test'],
                                      stdout=cls.NULL,
                                      stderr=cls.NULL)

    def test_model_output_and_consistency(self):
        self.models = glob.glob(os.path.join(self.settings_1['MODEL_PATH'],
                                             'batch*'))
        self.assertEqual(len(self.models), 14)

        target_models_1 = glob.glob(os.path.join(self.settings_1['MODEL_PATH'],
                                                 self.settings_1['RUN_NAME']+\
                                                        '_model_for_*.model'))

        target_models_2 = glob.glob(os.path.join(self.settings_2['MODEL_PATH'],
                                                 self.settings_2['RUN_NAME']+\
                                                        '_model_for_*.model'))

        target_models = target_models_1 + target_models_2

        self.assertEqual(set(self.models), set(target_models))

        random_subj = random.choice(self.settings_2['SUBJECTS'])

        model_1 = os.path.join(\
                self.settings_1['MODEL_PATH'],
                        'batch_test1_model_for_{0}'
                        '_using_{1}_feats.model'.format(random_subj,
                                                        self.settings_1['VERSION']))

        model_2 = os.path.join(\
                self.settings_2['MODEL_PATH'],
                        'batch_test2_model_for_{0}'
                        '_using_{1}_feats.model'.format(random_subj,
                                                        self.settings_2['VERSION']))


        #self.assertEqual(os.stat(model_1).st_size,
        #                 os.stat(model_2).st_size)

        for f in self.models:
            os.unlink(f)


    def test_submission_output_and_consistency(self):
        self.outputs = glob.glob(os.path.join(self.settings_2['SUBMISSION_PATH'],
                                              'batch*'))
        self.assertEqual(len(self.outputs), 2)

        with open(self.outputs[0], 'r') as output_1_fh:
            output_1 = set(output_1_fh.read().split('\n'))

        with open(self.outputs[1], 'r') as output_2_fh:
            output_2 = set(output_2_fh.read().split('\n'))
<commit_msg>Added comment explaining why I have removed model file size assertion as they differ slightly<commit_after>#!/usr/bin/env python3

import random
import copy
import numpy as np
import sklearn.pipeline
import json
import unittest
import glob
import warnings
import subprocess
import os
import python.utils as utils
import csv
import h5py


class testClassifieroptions(unittest.TestCase):
    pass


class testHDF5parsing(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        cls.settings_fh = 'test_settings.json'
        cls.settings = utils.get_settings(cls.settings_fh)
        cls.all_subjects = set(['Dog_1',
                             'Dog_2',
                             'Dog_3',
                             'Dog_4',
                             'Dog_5',
                             'Patient_1',
                             'Patient_2'])
        cls.all_types = set(['preictal',
                              'test',
                              'pseudointerictal',
                              'interictal',
                              'pseudopreictal'])

        cls.malformed_feat = 'malformed_feat'
        cls.malformed_file = os.path.join(cls.settings['TRAIN_DATA_PATH'],
                                          "{0}{1}.h5".format(
                                                    cls.malformed_feat,
                                                    cls.settings['VERSION']))

        cls.malformed_feat = h5py.File(cls.malformed_file, 'w')
        cls.malformed_feat.create_dataset('malfie', (10,10))
        cls.malformed_feat.close()

    def test_ioerror_warning(self):
        '''
        Assert a non-existent file correctly raises warning
        '''

        non_existent_feat = 'fake_feat'
        h5_file_name = os.path.join(self.settings['TRAIN_DATA_PATH'],
                                    "{0}{1}.h5".format(non_existent_feat,
                                                       self.settings['VERSION']))

        with warnings.catch_warnings(record=True) as w:
                dummy = utils.parse_matlab_HDF5(non_existent_feat,
                                                self.settings)

                self.assertEqual(len(w), 1, msg="Check there is one and only "
                                                "one warning raised")
                self.assertIs(w[-1].category, UserWarning, msg="Check that "
                                                                "warning raised "
                                                                "is a UserWarning ")
                self.assertEqual(str(w[-1].message),
                                 "{0} does not exist (or is not "
                                 "readable)".format(h5_file_name), msg="Check the "
                                                                        "warning is "
                                                                        "the correct "
                                                                        "format ")

    def test_parse_error_warning(self):
        '''
        Assert malformed HDF5 raises proper warning
        '''


        malformed_feat = 'malformed_feat'
        h5_file_name = os.path.join(self.settings['TRAIN_DATA_PATH'],
                                    "{0}{1}.h5".format(malformed_feat,
                                                       self.settings['VERSION']))

        with warnings.catch_warnings(record=True) as w:

                dummy = utils.parse_matlab_HDF5(malformed_feat,
                                                self.settings)

                self.assertEqual(len(w), 1, msg="Check one and only one error "
                                                "raised")
                self.assertIs(w[-1].category, UserWarning, msg="Check error is "
                                                                "UserWarning")
                self.assertEqual(str(w[-1].message), "Unable to "
                                                     "parse {0}".format(\
                                                                h5_file_name),
                                 msg="Check the warning raised is correct format")


    def test_hdf5_parse(self):
        '''
        Very basic check of parsed object structure containing minimum 3 dicts
        '''
        parsed_HDF5 = utils.parse_matlab_HDF5(self.settings['FEATURES'][0],
                                              self.settings)

        subjects = set(parsed_HDF5.keys())
        self.assertEqual(subjects,
                         self.all_subjects,
                         msg="Check that parsed HDF5 contains all subjects")

        typs = set(parsed_HDF5['Dog_1'].keys())
        self.assertEqual(typs,
                         self.all_types,
                         msg="Check that all ictypes are in parsed file by "
                             "checking the ictypes under Dog_1")

        num_interictal_dog1_segs = len(\
                parsed_HDF5['Dog_1']['interictal'].keys())

        self.assertEqual(num_interictal_dog1_segs,
                         480,
                         msg="Check there is the correct number of segments in "
                             "parsed HDF5s by checking dog1 interictal")

    @classmethod
    def tearDownClass(cls):
        os.unlink(cls.malformed_file)

class testBatchParallel(unittest.TestCase):
    '''
    Class to test the batch parallel script as well as consistency in output
    '''
    @classmethod
    def setUpClass(cls):
        cls.settings_fh_1 = os.path.join('batch_test', 'batch_test1.json')
        cls.settings_fh_2 = os.path.join('batch_test', 'batch_test2.json')
        cls.settings_1 = utils.get_settings(cls.settings_fh_1)
        cls.settings_2 = utils.get_settings(cls.settings_fh_2)

        cls.NULL = open(os.devnull, 'w')
        cls.proc = subprocess.call(['./parallel_batch_run.py',
                                      '-s', 'batch_test'],
                                      stdout=cls.NULL,
                                      stderr=cls.NULL)

    def test_model_output_and_consistency(self):
        self.models = glob.glob(os.path.join(self.settings_1['MODEL_PATH'],
                                             'batch*'))
        self.assertEqual(len(self.models), 14)

        target_models_1 = glob.glob(os.path.join(self.settings_1['MODEL_PATH'],
                                                 self.settings_1['RUN_NAME']+\
                                                        '_model_for_*.model'))

        target_models_2 = glob.glob(os.path.join(self.settings_2['MODEL_PATH'],
                                                 self.settings_2['RUN_NAME']+\
                                                        '_model_for_*.model'))

        target_models = target_models_1 + target_models_2

        self.assertEqual(set(self.models), set(target_models))

        random_subj = random.choice(self.settings_2['SUBJECTS'])

        model_1 = os.path.join(\
                self.settings_1['MODEL_PATH'],
                        'batch_test1_model_for_{0}'
                        '_using_{1}_feats.model'.format(random_subj,
                                                        self.settings_1['VERSION']))

        model_2 = os.path.join(\
                self.settings_2['MODEL_PATH'],
                        'batch_test2_model_for_{0}'
                        '_using_{1}_feats.model'.format(random_subj,
                                                        self.settings_2['VERSION']))

        # binaries different sizes - unsure why
        #self.assertEqual(os.stat(model_1).st_size,
        #                 os.stat(model_2).st_size)

        for f in self.models:
            os.unlink(f)


    def test_submission_output_and_consistency(self):
        self.outputs = glob.glob(os.path.join(self.settings_2['SUBMISSION_PATH'],
                                              'batch*'))
        self.assertEqual(len(self.outputs), 2)

        with open(self.outputs[0], 'r') as output_1_fh:
            output_1 = set(output_1_fh.read().split('\n'))

        with open(self.outputs[1], 'r') as output_2_fh:
            output_2 = set(output_2_fh.read().split('\n'))

{'fraction_non_alphanumeric': 0.09402390438247012, 'fraction_numerical': 0.012503830830524058, 'mean_word_length': 1.2283529090412455, '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': 128, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4780135', 'n_tokens_mistral': 4114, 'n_tokens_neox': 3943, 'n_words': 711}