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

"""
@package ion.util.test.test_direct_coverage_utils
@file ion/util/test/test_direct_coverage_utils.py
@author <NAME>
@brief 
"""

import os
import unittest
import numpy as np
from nose.plugins.attrib import attr

from coverage_model import ParameterContext, SparseConstantType, BooleanType
from ion.services.dm.test.dm_test_case import DMTestCase, Streamer
from ion.services.dm.test.test_dm_end_2_end import DatasetMonitor
from ion.util.direct_coverage_utils import DirectCoverageAccess
from pyon.ion.resource import PRED


from subprocess import call
import re
not_have_h5stat = call('which h5stat'.split(), stdout=open('/dev/null','w'))
if not not_have_h5stat:
    from subprocess import check_output
    from distutils.version import StrictVersion
    output = check_output('h5stat -V'.split())
    version_str = re.match(r'.*(\d+\.\d+\.\d+).*', output).groups()[0]
    h5stat_correct_version = StrictVersion(version_str) >= StrictVersion('1.8.9')


@attr('INT', group='dm')
class TestDirectCoverageAccess(DMTestCase):

    def use_monitor(self, dataset_id, samples=10, wait=10):
        for x in xrange(samples):
            monitor = DatasetMonitor(dataset_id)
            monitor.event.wait(wait)
            monitor.stop()

    def make_cal_data_product(self):
        # Get a precompiled parameter dictionary with basic ctd fields
        pdict_id = self.dataset_management.read_parameter_dictionary_by_name('ctd_parsed_param_dict',id_only=True)
        context_ids = self.dataset_management.read_parameter_contexts(pdict_id, id_only=True)

        # Add a handful of Calibration Coefficient parameters
        for cc in ['cc_ta0', 'cc_ta1', 'cc_ta2', 'cc_ta3', 'cc_toffset']:
            c = ParameterContext(cc, param_type=SparseConstantType(value_encoding='float32', fill_value=-9999))
            c.uom = '1'
            context_ids.append(self.dataset_management.create_parameter_context(cc, c.dump()))

        pdict_id = self.dataset_management.create_parameter_dictionary('calcoeff_dict', context_ids, temporal_context='time')

        data_product_id = self.create_data_product('calcoeff_dp', pdict_id=pdict_id)
        self.activate_data_product(data_product_id)
        dataset_id, _ = self.resource_registry.find_objects(data_product_id, PRED.hasDataset, id_only=True)

        return data_product_id, dataset_id[0]

    def make_manual_upload_data_product(self):
        # Get a precompiled parameter dictionary with basic ctd fields
        pdict_id = self.dataset_management.read_parameter_dictionary_by_name('ctd_parsed_param_dict',id_only=True)
        context_ids = self.dataset_management.read_parameter_contexts(pdict_id, id_only=True)

        # Add a handful of Calibration Coefficient parameters
        for cc in ['temp_hitl_qc', 'cond_hitl_qc']:
            c = ParameterContext(cc, param_type=BooleanType())
            c.uom = '1'
            context_ids.append(self.dataset_management.create_parameter_context(cc, c.dump()))

        pdict_id = self.dataset_management.create_parameter_dictionary('manup_dict', context_ids, temporal_context='time')

        data_product_id = self.create_data_product('manup_dp', pdict_id=pdict_id)
        self.activate_data_product(data_product_id)
        dataset_id, _ = self.resource_registry.find_objects(data_product_id, PRED.hasDataset, id_only=True)

        return data_product_id, dataset_id[0]

    def make_ctd_data_product(self):
        data_product_id = self.create_data_product('test_prod', param_dict_name='ctd_parsed_param_dict')
        self.activate_data_product(data_product_id)
        dataset_id, _ = self.resource_registry.find_objects(data_product_id, PRED.hasDataset, id_only=True)

        return data_product_id, dataset_id[0]

    def test_dca_ingestion_pause_resume(self):
        data_product_id, dataset_id = self.make_ctd_data_product()

        streamer = Streamer(data_product_id, interval=1)
        self.addCleanup(streamer.stop)

        # Let a couple samples accumulate
        self.use_monitor(dataset_id, samples=2)

        # Go into DCA and get an editable handle to the coverage
        with DirectCoverageAccess() as dca:
            with dca.get_editable_coverage(dataset_id) as cov: # <-- This pauses ingestion
                monitor = DatasetMonitor(dataset_id)
                monitor.event.wait(7) # <-- ~7 Samples should accumulate on the ingestion queue
                self.assertFalse(monitor.event.is_set()) # Verifies that nothing was processed (i.e. ingestion is actually paused)
                monitor.stop()

        # Stop the streamer
        streamer.stop()

        cont = True
        while cont:
            monitor = DatasetMonitor(dataset_id)
            if not monitor.event.wait(10):
                cont = False
            monitor.stop()

        with DirectCoverageAccess() as dca:
            with dca.get_read_only_coverage(dataset_id) as cov:
                self.assertGreaterEqual(cov.num_timesteps, 9)

    def test_dca_coverage_reuse(self):
        data_product_id, dataset_id = self.make_ctd_data_product()

        streamer = Streamer(data_product_id, interval=1)
        self.addCleanup(streamer.stop)

        # Let a couple samples accumulate
        self.use_monitor(dataset_id, samples=2)

        with DirectCoverageAccess() as dca:
            with dca.get_read_only_coverage(dataset_id) as cov:
                self.assertFalse(cov.closed)

            self.assertTrue(cov.closed)

            with dca.get_editable_coverage(dataset_id) as cov:
                self.assertFalse(cov.closed)

            self.assertTrue(cov.closed)

            with dca.get_read_only_coverage(dataset_id) as cov:
                self.assertFalse(cov.closed)

            self.assertTrue(cov.closed)

    @unittest.skipIf(os.getenv('CEI_LAUNCH_TEST', False), 'Host requires file-system access to coverage files, CEI mode does not support.')
    def test_upload_calibration_coefficients(self):
        data_product_id, dataset_id = self.make_cal_data_product()

        streamer = Streamer(data_product_id, interval=0.5)
        self.addCleanup(streamer.stop)

        # Let at least 10 samples accumulate
        self.use_monitor(dataset_id, samples=10)

        # Verify that the CC parameters are fill value
        with DirectCoverageAccess() as dca:
            with dca.get_read_only_coverage(dataset_id) as cov:
                for p in [p for p in cov.list_parameters() if p.startswith('cc_')]:
                    np.testing.assert_equal(cov.get_parameter_values(p, -1), -9999.)

        # Upload the calibration coefficients - this pauses ingestion, performs the upload, and resumes ingestion
        with DirectCoverageAccess() as dca:
            dca.upload_calibration_coefficients(dataset_id, 'test_data/testcalcoeff.csv', 'test_data/testcalcoeff.yml')

        # Let a little more data accumulate
        self.use_monitor(dataset_id, samples=2)

        # Verify that the CC parameters now have the correct values
        want_vals = {
            'cc_ta0': np.float32(1.155787e-03),
            'cc_ta1': np.float32(2.725208e-04),
            'cc_ta2': np.float32(-7.526811e-07),
            'cc_ta3': np.float32(1.716270e-07),
            'cc_toffset': np.float32(0.000000e+00)
        }
        with DirectCoverageAccess() as dca:
            with dca.get_read_only_coverage(dataset_id) as cov:
                for p in [p for p in cov.list_parameters() if p.startswith('cc_')]:
                    np.testing.assert_equal(cov.get_parameter_values(p, -1), want_vals[p])

    @unittest.skipIf(os.getenv('CEI_LAUNCH_TEST', False), 'Host requires file-system access to coverage files, CEI mode does not support.')
    def test_manual_data_upload(self):
        data_product_id, dataset_id = self.make_manual_upload_data_product()

        streamer = Streamer(data_product_id, interval=0.5, simple_time=True)
        self.addCleanup(streamer.stop)<commit_msg>Makes all DCA tests 'LOCOINT'<commit_after>#!/usr/bin/env python

"""
@package ion.util.test.test_direct_coverage_utils
@file ion/util/test/test_direct_coverage_utils.py
@author <NAME>
@brief 
"""

import os
import unittest
import numpy as np
from nose.plugins.attrib import attr

from coverage_model import ParameterContext, SparseConstantType, BooleanType
from ion.services.dm.test.dm_test_case import DMTestCase, Streamer
from ion.services.dm.test.test_dm_end_2_end import DatasetMonitor
from ion.util.direct_coverage_utils import DirectCoverageAccess
from pyon.ion.resource import PRED


from subprocess import call
import re
not_have_h5stat = call('which h5stat'.split(), stdout=open('/dev/null','w'))
if not not_have_h5stat:
    from subprocess import check_output
    from distutils.version import StrictVersion
    output = check_output('h5stat -V'.split())
    version_str = re.match(r'.*(\d+\.\d+\.\d+).*', output).groups()[0]
    h5stat_correct_version = StrictVersion(version_str) >= StrictVersion('1.8.9')


@attr('INT', group='dm')
class TestDirectCoverageAccess(DMTestCase):

    def use_monitor(self, dataset_id, samples=10, wait=10):
        for x in xrange(samples):
            monitor = DatasetMonitor(dataset_id)
            monitor.event.wait(wait)
            monitor.stop()

    def make_cal_data_product(self):
        # Get a precompiled parameter dictionary with basic ctd fields
        pdict_id = self.dataset_management.read_parameter_dictionary_by_name('ctd_parsed_param_dict',id_only=True)
        context_ids = self.dataset_management.read_parameter_contexts(pdict_id, id_only=True)

        # Add a handful of Calibration Coefficient parameters
        for cc in ['cc_ta0', 'cc_ta1', 'cc_ta2', 'cc_ta3', 'cc_toffset']:
            c = ParameterContext(cc, param_type=SparseConstantType(value_encoding='float32', fill_value=-9999))
            c.uom = '1'
            context_ids.append(self.dataset_management.create_parameter_context(cc, c.dump()))

        pdict_id = self.dataset_management.create_parameter_dictionary('calcoeff_dict', context_ids, temporal_context='time')

        data_product_id = self.create_data_product('calcoeff_dp', pdict_id=pdict_id)
        self.activate_data_product(data_product_id)
        dataset_id, _ = self.resource_registry.find_objects(data_product_id, PRED.hasDataset, id_only=True)

        return data_product_id, dataset_id[0]

    def make_manual_upload_data_product(self):
        # Get a precompiled parameter dictionary with basic ctd fields
        pdict_id = self.dataset_management.read_parameter_dictionary_by_name('ctd_parsed_param_dict',id_only=True)
        context_ids = self.dataset_management.read_parameter_contexts(pdict_id, id_only=True)

        # Add a handful of Calibration Coefficient parameters
        for cc in ['temp_hitl_qc', 'cond_hitl_qc']:
            c = ParameterContext(cc, param_type=BooleanType())
            c.uom = '1'
            context_ids.append(self.dataset_management.create_parameter_context(cc, c.dump()))

        pdict_id = self.dataset_management.create_parameter_dictionary('manup_dict', context_ids, temporal_context='time')

        data_product_id = self.create_data_product('manup_dp', pdict_id=pdict_id)
        self.activate_data_product(data_product_id)
        dataset_id, _ = self.resource_registry.find_objects(data_product_id, PRED.hasDataset, id_only=True)

        return data_product_id, dataset_id[0]

    def make_ctd_data_product(self):
        data_product_id = self.create_data_product('test_prod', param_dict_name='ctd_parsed_param_dict')
        self.activate_data_product(data_product_id)
        dataset_id, _ = self.resource_registry.find_objects(data_product_id, PRED.hasDataset, id_only=True)

        return data_product_id, dataset_id[0]

    @attr('LOCOINT')
    def test_dca_ingestion_pause_resume(self):
        data_product_id, dataset_id = self.make_ctd_data_product()

        streamer = Streamer(data_product_id, interval=1)
        self.addCleanup(streamer.stop)

        # Let a couple samples accumulate
        self.use_monitor(dataset_id, samples=2)

        # Go into DCA and get an editable handle to the coverage
        with DirectCoverageAccess() as dca:
            with dca.get_editable_coverage(dataset_id) as cov: # <-- This pauses ingestion
                monitor = DatasetMonitor(dataset_id)
                monitor.event.wait(7) # <-- ~7 Samples should accumulate on the ingestion queue
                self.assertFalse(monitor.event.is_set()) # Verifies that nothing was processed (i.e. ingestion is actually paused)
                monitor.stop()

        # Stop the streamer
        streamer.stop()

        cont = True
        while cont:
            monitor = DatasetMonitor(dataset_id)
            if not monitor.event.wait(10):
                cont = False
            monitor.stop()

        with DirectCoverageAccess() as dca:
            with dca.get_read_only_coverage(dataset_id) as cov:
                self.assertGreaterEqual(cov.num_timesteps, 9)

    @attr('LOCOINT')
    def test_dca_coverage_reuse(self):
        data_product_id, dataset_id = self.make_ctd_data_product()

        streamer = Streamer(data_product_id, interval=1)
        self.addCleanup(streamer.stop)

        # Let a couple samples accumulate
        self.use_monitor(dataset_id, samples=2)

        with DirectCoverageAccess() as dca:
            with dca.get_read_only_coverage(dataset_id) as cov:
                self.assertFalse(cov.closed)

            self.assertTrue(cov.closed)

            with dca.get_editable_coverage(dataset_id) as cov:
                self.assertFalse(cov.closed)

            self.assertTrue(cov.closed)

            with dca.get_read_only_coverage(dataset_id) as cov:
                self.assertFalse(cov.closed)

            self.assertTrue(cov.closed)

    @attr('LOCOINT')
    @unittest.skipIf(os.getenv('CEI_LAUNCH_TEST', False), 'Host requires file-system access to coverage files, CEI mode does not support.')
    def test_upload_calibration_coefficients(self):
        data_product_id, dataset_id = self.make_cal_data_product()

        streamer = Streamer(data_product_id, interval=0.5)
        self.addCleanup(streamer.stop)

        # Let at least 10 samples accumulate
        self.use_monitor(dataset_id, samples=10)

        # Verify that the CC parameters are fill value
        with DirectCoverageAccess() as dca:
            with dca.get_read_only_coverage(dataset_id) as cov:
                for p in [p for p in cov.list_parameters() if p.startswith('cc_')]:
                    np.testing.assert_equal(cov.get_parameter_values(p, -1), -9999.)

        # Upload the calibration coefficients - this pauses ingestion, performs the upload, and resumes ingestion
        with DirectCoverageAccess() as dca:
            dca.upload_calibration_coefficients(dataset_id, 'test_data/testcalcoeff.csv', 'test_data/testcalcoeff.yml')

        # Let a little more data accumulate
        self.use_monitor(dataset_id, samples=2)

        # Verify that the CC parameters now have the correct values
        want_vals = {
            'cc_ta0': np.float32(1.155787e-03),
            'cc_ta1': np.float32(2.725208e-04),
            'cc_ta2': np.float32(-7.526811e-07),
            'cc_ta3': np.float32(1.716270e-07),
            'cc_toffset': np.float32(0.000000e+00)
        }
        with DirectCoverageAccess() as dca:
            with dca.get_read_only_coverage(dataset_id) as cov:
                for p in [p for p in cov.list_parameters() if p.startswith('cc_')]:
                    np.testing.assert_equal(cov.get_parameter_values(p, -1), want_vals[p])

    @attr('LOCOINT')
    @unittest.skipIf(os.getenv('CEI_LAUNCH_TEST', False), 'Host requires file-system access to coverage files, CEI mode does not support.')
    def test_manual_data_upload(self):
        data_product_id, dataset_id = self.make_manual_upload_data_product()

        streamer = Streamer(data_product_id, interval=0.5, simple_time=True)
        self.addCleanup(streamer.stop)
{'fraction_non_alphanumeric': 0.09255558331251561, 'fraction_numerical': 0.013989507869098177, 'mean_word_length': 3.2418543046357615, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, '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': '1073681', 'n_tokens_mistral': 5177, 'n_tokens_neox': 4921, 'n_words': 981}