<commit_before>import csv
import datetime
import os
import unittest
import pandas as pd
import numpy as np
import h5py
from lxml import etree
import sys
sys.path.append('/usr/local/project/flux-python-api/')
from fluxpy.legacy.transform import bulk_hdf5_to_csv
from fluxpy.mediators import Grid4DMediator, Grid3DMediator, Unstructured3DMediator
from fluxpy.models import KrigedXCO2Matrix, XCO2Matrix, SpatioTemporalMatrix
from fluxpy.colors import DivergingColors

class TestSpatioTemporalMatrixes(unittest.TestCase):
    '''Tests for proper handling of inverted CO2 surface fluxes (e.g. CASA GFED output)'''

    mediator = Grid4DMediator()
    path = '/usr/local/project/flux-python-api/fluxpy/tests/'

    def test_model_instance(self):
        '''Should properly instantiate an SpatioTemporalMatrix model instance'''
        flux = SpatioTemporalMatrix(os.path.join(self.path, 'casagfed2004.mat'),
            timestamp='2004-06-30T00:00:00', var_name='test', span=10800)

        self.assertEqual(flux.var_name, 'test')
        self.assertEqual(flux.step, 10800)
        self.assertEqual(flux.span, 10800)
        self.assertEqual(flux.timestamp, '2004-06-30T00:00:00')

    def test_model_var_name_inference(self):
        '''Should infer the var_name in an SpatioTemporalMatrix model instance'''
        flux = SpatioTemporalMatrix(os.path.join(self.path, 'casagfed2004.mat'))

        self.assertEqual(flux.var_name, 'test')

    def test_model_describe(self):
        '''Should produce metadata for a SpatioTemporalMatrix model instance'''
        flux = SpatioTemporalMatrix(os.path.join(self.path, 'casagfed2004.mat'),
            timestamp='2004-06-30T00:00:00', var_name='test')
        
        df = flux.describe()
        self.assertEqual(df['bbox'], (-166.5, 60.5, -163.5, 68.5))
        self.assertEqual(df['bboxmd5'], '51d5738489b4ae4fa8623f867de527ce')
        self.assertEqual(df['dates'], ['2004-06-30T00:00:00', '2004-06-30T21:00:00'])
        self.assertEqual(df['gridded'], True)
        self.assertEqual(df['gridres'], {'units': 'degrees', 'x': 1.0, 'y': 1.0})
        self.assertEqual(df['spans'], [10800])
        self.assertEqual(df['steps'], [10800])
    
    def test_model_extract(self):
        '''Should extract a DataFrame in an SpatioTemporalMatrix model instance'''
        flux = SpatioTemporalMatrix(os.path.join(self.path, 'casagfed2004.mat'),
            timestamp='2004-06-30T00:00:00', var_name='test')

        df = flux.extract()
        self.assertEqual(df.shape, (10, 8))
        self.assertEqual(str(df.columns[1]), '2004-06-30 03:00:00')
        self.assertEqual(df.index.values[1], (-165.5, 61.5))

    def test_save_to_db(self):
        '''Should successfully save proper data representation to database'''
        flux = SpatioTemporalMatrix(os.path.join(self.path, 'casagfed2004.mat'),
            timestamp='2004-06-30T00:00:00', var_name='test')

        self.mediator.save('test3', flux)
        query = self.mediator.client[self.mediator.db_name]['test3'].find({
            '_id': datetime.datetime(2004, 6, 30, 0, 0, 0),
        })
        self.assertEqual(len(query[0]['values']), 10)
        self.assertEqual(query[0]['values'][0], 0.08)

        # Drop the old collection; it will be recreated when inserting
        self.mediator.client[self.mediator.db_name].drop_collection('test3')
        self.mediator.client[self.mediator.db_name]['coord_index'].remove({
            '_id': 'test3'
        })
        self.mediator.client[self.mediator.db_name]['metadata'].remove({
            '_id': 'test3'
        })


class TestXCO2Data(unittest.TestCase):
    '''Tests for proper handling of XCO2 retrievals'''

    mediator = Unstructured3DMediator()
    path = '/usr/local/project/flux-python-api/fluxpy/tests/'

    def test_model_instance(self):
        '''Should properly instantiate a model instance'''
        xco2 = XCO2Matrix(os.path.join(self.path, 'xco2.mat'),
            timestamp='2009-06-15')

        self.assertEqual(xco2.var_name, 'XCO2')
        self.assertEqual(xco2.step, 86400)
        self.assertEqual(xco2.span, None)
        self.assertEqual(xco2.timestamp, '2009-06-15')
        
    def test_model_extract(self):
        '''Should create proper DataFrame from reading file data'''
        xco2 = XCO2Matrix(os.path.join(self.path, 'xco2.mat'),
            timestamp='2009-06-15')

        df1 = xco2.extract()
        self.assertEqual(df1.shape, (1311, 7))

        # Should allow overrides in the extract() method
        df2 = xco2.extract(timestamp='2010-01-01')
        self.assertEqual(xco2.timestamp, '2010-01-01')
            
    def test_save_to_db(self):
        '''Should successfully save proper data representation to database'''
        xco2 = XCO2Matrix(os.path.join(self.path, 'xco2.mat'),
            timestamp='2009-06-15')

        self.mediator.save('test', xco2)
        query = self.mediator.client[self.mediator.db_name]['test'].find({
            'timestamp': datetime.datetime(2009, 6, 16, 0, 0, 0),
        })
        self.assertEqual(query[0]['value'], 386.79)

        # Drop the old collection; it will be recreated when inserting
        self.mediator.client[self.mediator.db_name].drop_collection('test')
        self.mediator.client[self.mediator.db_name]['coord_index'].remove({
            '_id': 'test'
        })
        self.mediator.client[self.mediator.db_name]['metadata'].remove({
            '_id': 'test'
        })
        

class TestKrigedXCO2Data(unittest.TestCase):
    '''Tests for proper handling of kriged (gridded) XCO2 data'''

    mediator = Grid3DMediator()
    path = '/usr/local/project/flux-python-api/fluxpy/tests/'

    def test_model_instance(self):
        '''Should properly instantiate a model instance'''
        xco2 = KrigedXCO2Matrix(os.path.join(self.path, 'kriged_xco2.mat'),
            timestamp='2009-06-15')

        self.assertEqual(xco2.var_name, 'krigedData')
        self.assertEqual(xco2.step, None)
        self.assertEqual(xco2.span, 518400)
        self.assertEqual(xco2.timestamp, '2009-06-15')
        
    def test_model_extract(self):
        '''Should create proper DataFrame from reading file data'''
        xco2 = KrigedXCO2Matrix(os.path.join(self.path, 'kriged_xco2.mat'),
            timestamp='2009-06-15')

        df1 = xco2.extract()
        self.assertEqual(df1.shapeHu, (14210, 9))

        # Should allow overrides in the extract() method
        df2 = xco2.extract(timestamp='2010-01-01')
        self.assertEqual(xco2.timestamp, '2010-01-01')
            
    def test_save_to_db(self):
        '''Should successfully save proper data representation to database'''
        xco2 = KrigedXCO2Matrix(os.path.join(self.path, 'kriged_xco2.mat'),
            timestamp='2009-06-15')

        self.mediator.save('test2', xco2)<commit_msg>Fixes a few minor errors in tests.py that were causing unwarranted test failures<commit_after>import csv
import datetime
import os
import unittest
import pandas as pd
import numpy as np
import h5py
from lxml import etree
import sys
sys.path.append('/usr/local/project/flux-python-api/')
from fluxpy.legacy.transform import bulk_hdf5_to_csv
from fluxpy.mediators import Grid4DMediator, Grid3DMediator, Unstructured3DMediator
from fluxpy.models import KrigedXCO2Matrix, XCO2Matrix, SpatioTemporalMatrix
from fluxpy.colors import DivergingColors

class TestSpatioTemporalMatrixes(unittest.TestCase):
    '''Tests for proper handling of inverted CO2 surface fluxes (e.g. CASA GFED output)'''

    mediator = Grid4DMediator()
    path = '/usr/local/project/flux-python-api/fluxpy/tests/'

    def test_model_instance(self):
        '''Should properly instantiate an SpatioTemporalMatrix model instance'''
        flux = SpatioTemporalMatrix(os.path.join(self.path, 'casagfed2004.mat'),
            timestamp='2004-06-30T00:00:00', var_name='test', span=10800)

        self.assertEqual(flux.var_name, 'test')
        self.assertEqual(flux.step, 10800)
        self.assertEqual(flux.span, 10800)
        self.assertEqual(flux.timestamp, '2004-06-30T00:00:00')

    def test_model_var_name_inference(self):
        '''Should infer the var_name in an SpatioTemporalMatrix model instance'''
        flux = SpatioTemporalMatrix(os.path.join(self.path, 'casagfed2004.mat'))

        self.assertEqual(flux.var_name, 'test')

    def test_model_describe(self):
        '''Should produce metadata for a SpatioTemporalMatrix model instance'''
        flux = SpatioTemporalMatrix(os.path.join(self.path, 'casagfed2004.mat'),
            timestamp='2004-06-30T00:00:00', var_name='test',span=10800)
        
        df = flux.describe()
        self.assertEqual(df['bbox'], (-166.5, 60.5, -163.5, 68.5))
        self.assertEqual(df['bboxmd5'], '51d5738489b4ae4fa8623f867de527ce')
        self.assertEqual(df['dates'], ['2004-06-30T00:00:00', '2004-06-30T21:00:00'])
        self.assertEqual(df['gridded'], True)
        self.assertEqual(df['gridres'], {'units': 'degrees', 'x': 1.0, 'y': 1.0})
        self.assertEqual(df['spans'], [10800])
        self.assertEqual(df['steps'], [10800])
    
    def test_model_extract(self):
        '''Should extract a DataFrame in an SpatioTemporalMatrix model instance'''
        flux = SpatioTemporalMatrix(os.path.join(self.path, 'casagfed2004.mat'),
            timestamp='2004-06-30T00:00:00', var_name='test')

        df = flux.extract()
        self.assertEqual(df.shape, (10, 8))
        self.assertEqual(str(df.columns[1]), '2004-06-30 03:00:00')
        self.assertEqual(df.index.values[1], (-165.5, 61.5))

    def test_save_to_db(self):
        '''Should successfully save proper data representation to database'''
        flux = SpatioTemporalMatrix(os.path.join(self.path, 'casagfed2004.mat'),
            timestamp='2004-06-30T00:00:00', var_name='test')

        self.mediator.save('test3', flux)
        query = self.mediator.client[self.mediator.db_name]['test3'].find({
            '_id': datetime.datetime(2004, 6, 30, 0, 0, 0),
        })
        self.assertEqual(len(query[0]['values']), 10)
        self.assertEqual(query[0]['values'][0], 0.08)

        # Drop the old collection; it will be recreated when inserting
        self.mediator.client[self.mediator.db_name].drop_collection('test3')
        self.mediator.client[self.mediator.db_name]['coord_index'].remove({
            '_id': 'test3'
        })
        self.mediator.client[self.mediator.db_name]['metadata'].remove({
            '_id': 'test3'
        })


class TestXCO2Data(unittest.TestCase):
    '''Tests for proper handling of XCO2 retrievals'''

    mediator = Unstructured3DMediator()
    path = '/usr/local/project/flux-python-api/fluxpy/tests/'

    def test_model_instance(self):
        '''Should properly instantiate a model instance'''
        xco2 = XCO2Matrix(os.path.join(self.path, 'xco2.mat'),
            timestamp='2009-06-15')

        self.assertEqual(xco2.var_name, 'XCO2')
        self.assertEqual(xco2.step, 86400)
        self.assertEqual(xco2.span, None)
        self.assertEqual(xco2.timestamp, '2009-06-15')
        
    def test_model_extract(self):
        '''Should create proper DataFrame from reading file data'''
        xco2 = XCO2Matrix(os.path.join(self.path, 'xco2.mat'),
            timestamp='2009-06-15')

        df1 = xco2.extract()
        self.assertEqual(df1.shape, (1311, 7))

        # Should allow overrides in the extract() method
        df2 = xco2.extract(timestamp='2010-01-01')
        self.assertEqual(xco2.timestamp, '2010-01-01')
            
    def test_save_to_db(self):
        '''Should successfully save proper data representation to database'''
        xco2 = XCO2Matrix(os.path.join(self.path, 'xco2.mat'),
            timestamp='2009-06-15')

        self.mediator.save('test', xco2)
        query = self.mediator.client[self.mediator.db_name]['test'].find({
            'timestamp': datetime.datetime(2009, 6, 16, 0, 0, 0),
        })
        self.assertEqual(query[0]['value'], 386.79)

        # Drop the old collection; it will be recreated when inserting
        self.mediator.client[self.mediator.db_name].drop_collection('test')
        self.mediator.client[self.mediator.db_name]['coord_index'].remove({
            '_id': 'test'
        })
        self.mediator.client[self.mediator.db_name]['metadata'].remove({
            '_id': 'test'
        })
        

class TestKrigedXCO2Data(unittest.TestCase):
    '''Tests for proper handling of kriged (gridded) XCO2 data'''

    mediator = Grid3DMediator()
    path = '/usr/local/project/flux-python-api/fluxpy/tests/'

    def test_model_instance(self):
        '''Should properly instantiate a model instance'''
        xco2 = KrigedXCO2Matrix(os.path.join(self.path, 'kriged_xco2.mat'),
            timestamp='2009-06-15')

        self.assertEqual(xco2.var_name, 'krigedData')
        self.assertEqual(xco2.step, None)
        self.assertEqual(xco2.span, 518400)
        self.assertEqual(xco2.timestamp, '2009-06-15')
        
    def test_model_extract(self):
        '''Should create proper DataFrame from reading file data'''
        xco2 = KrigedXCO2Matrix(os.path.join(self.path, 'kriged_xco2.mat'),
            timestamp='2009-06-15')

        df1 = xco2.extract()
        self.assertEqual(df1.shape, (14210, 9))

        # Should allow overrides in the extract() method
        df2 = xco2.extract(timestamp='2010-01-01')
        self.assertEqual(xco2.timestamp, '2010-01-01')
            
    def test_save_to_db(self):
        '''Should successfully save proper data representation to database'''
        xco2 = KrigedXCO2Matrix(os.path.join(self.path, 'kriged_xco2.mat'),
            timestamp='2009-06-15')

        self.mediator.save('test2', xco2)
{'fraction_non_alphanumeric': 0.13805425904317387, 'fraction_numerical': 0.06206242707117853, 'mean_word_length': 3.0523049645390072, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '3354726', 'n_tokens_mistral': 4936, 'n_tokens_neox': 4366, 'n_words': 821}