path: root/src/argaze.test/DataFeatures.py

""" """

"""
This program is free software: you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by the Free Software
Foundation, either version 3 of the License, or (at your option) any later
version.
This program is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
You should have received a copy of the GNU General Public License along with
this program. If not, see <https://www.gnu.org/licenses/>.
"""

__author__ = "Théo de la Hogue"
__credits__ = []
__copyright__ = "Copyright 2023, Ecole Nationale de l'Aviation Civile (ENAC)"
__license__ = "GPLv3"

import unittest
from dataclasses import dataclass, field
import os

from argaze import DataFeatures

import pandas
import numpy

class BasicDataClass(DataFeatures.TimestampedObject):
    """Define a basic dataclass for testing purpose."""

    def __init__(self, value: tuple = (), message: str = None, **kwargs):

        DataFeatures.TimestampedObject.__init__(self, **kwargs)

        self.__value = value
        self.__message = message

    @property
    def value(self):
        return self.__value

    @property
    def message(self):
        return self.__message
    
def random_data_list(size):
    """ Generate a random TimestampedObjectsList for testing purpose.
        Timestamps are current time.
        Values are tuples containing an expected value and a random value.
    """
    import random
    import time

    data_list = []
    
    for i in range(0, size):

        # Edit data
        random_data = BasicDataClass((i, random.random()), f'test_{i}')
        
        # Timestamp data
        random_data.timestamp = time.time()

        # Store data
        data_list.append(random_data)

        time.sleep(0.0001)

    return DataFeatures.TimestampedObjectsList(BasicDataClass, data_list)

# DEBUG
'''
print('test_as_dataframe: export ts_data_file.json')
current_directory = os.path.dirname(os.path.abspath(__file__))
json_filepath = os.path.join(current_directory, 'utils/ts_data_file.json')
random_data_list(10).to_json(json_filepath)
'''

class TestTimestampedObjectsListClass(unittest.TestCase):
    """Test TimestampedObjectsList class."""

    def test_new(self):
        """Test TimestampedObjectsList creation."""
        
        # Check TimestampedObjectsList length after creation
        self.assertEqual(len(DataFeatures.TimestampedObjectsList(BasicDataClass)), 0)

        # Check TimestampedObjectsList timestamps after creation
        self.assertEqual(DataFeatures.TimestampedObjectsList(BasicDataClass).timestamps(), [])

        # Check TimestampedObjectsList items after creation
        self.assertEqual(DataFeatures.TimestampedObjectsList(BasicDataClass), [])

        # Check that TimestampedObjectsList creation fails when data are not timestamped
        with self.assertRaises(ValueError):

            data_list = [BasicDataClass((0, 0))]
            DataFeatures.TimestampedObjectsList(BasicDataClass, data_list)
        
    def test_as_dataframe(self):
        """Test TimestampedObjectsList as_dataframe method."""

        data_frame = random_data_list(10).as_dataframe()

        # Check dataframe conversion
        self.assertEqual(data_frame.index.name, "timestamp")
        self.assertEqual(data_frame.index.size, 10)
        self.assertEqual(data_frame.columns.size, 2)
        self.assertEqual(data_frame.index.dtype, 'float64')
        self.assertEqual(data_frame["value"].dtype, 'object')
        self.assertEqual(data_frame["message"].dtype, 'object')

        # Check data exclusion option
        data_frame = random_data_list(10).as_dataframe(exclude=["value"])

        self.assertEqual(data_frame.index.name, "timestamp")
        self.assertEqual(data_frame.index.size, 10)
        self.assertEqual(data_frame.columns.size, 1)

        # Check dataframe split option
        data_frame = random_data_list(10).as_dataframe(split={"value": ["value_0", "value_1"]})

        self.assertEqual(data_frame.index.name, "timestamp")
        self.assertEqual(data_frame.index.size, 10)
        self.assertEqual(data_frame.columns.size, 3)
        self.assertEqual(data_frame["value_0"].dtype, 'int64')
        self.assertEqual(data_frame["value_1"].dtype, 'float64')
    
    def test_from_dataframe(self):
        """Test TimestampedObjectsList creation from pandas dataframe."""

        data_frame = random_data_list(10).as_dataframe()

        # Check dataframe conversion
        data_list = DataFeatures.TimestampedObjectsList.from_dataframe(BasicDataClass, data_frame)

        self.assertEqual(len(data_list), 10)
    
    def test_from_json(self):
        """Test TimestampedObjectsList creation from json file."""

        # Edit dataframe csv file path
        current_directory = os.path.dirname(os.path.abspath(__file__))
        json_filepath = os.path.join(current_directory, 'utils/ts_data_file.json')

        # Load TimestampedObjectsList from json file
        data_list = DataFeatures.TimestampedObjectsList.from_json(BasicDataClass, json_filepath)

        self.assertEqual(len(data_list), 10)
        self.assertEqual(type(data_list[0]), BasicDataClass)
    
    def test___repr__(self):
        """Test TimestampedObjectsList string representation."""

        self.assertEqual(repr(DataFeatures.TimestampedObjectsList(BasicDataClass)), "[]")

        data_list = [BasicDataClass((0, 0), 'test', timestamp=0)]
        self.assertEqual(repr(DataFeatures.TimestampedObjectsList(BasicDataClass, data_list)), "[{\"value\": [0, 0], \"message\": \"test\", \"timestamp\": 0}]")

    def test___str__(self):
        """Test TimestampedObjectsList string representation."""

        self.assertEqual(str(DataFeatures.TimestampedObjectsList(BasicDataClass)), "[]")

        data_list = [BasicDataClass((0, 0), 'test', timestamp=0)]
        self.assertEqual(str(DataFeatures.TimestampedObjectsList(BasicDataClass, data_list)), "[{\"value\": [0, 0], \"message\": \"test\", \"timestamp\": 0}]")

    def test_append(self):
        """Test TimestampedObjectsList append method."""

        data_list = DataFeatures.TimestampedObjectsList(BasicDataClass)
        next_data = BasicDataClass((0, 0), 'test', timestamp=0)

        data_list.append(next_data)

        self.assertEqual(len(data_list), 1)
        self.assertEqual(data_list.timestamps(), [0])
    
    def test_pop_first(self):
        """Test TimestampedObjectsList pop_first method."""

        data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \
            [
                BasicDataClass(message="A", timestamp=0),
                BasicDataClass(message="B", timestamp=1)
            ])

        first = data_list.pop(0)

        self.assertEqual(len(data_list), 1)
        self.assertEqual(first.message, "A")
        self.assertEqual(first.timestamp, 0)
        self.assertEqual(data_list[0].message, "B")
        self.assertEqual(data_list[0].timestamp, 1)

    def test_pop_last_until(self):
        """Test TimestampedObjectsList pop_last_until method."""

        data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \
        [
            BasicDataClass(message="A", timestamp=0),
            BasicDataClass(message="B", timestamp=1),
            BasicDataClass(message="C", timestamp=2),
            BasicDataClass(message="D", timestamp=3)
        ])

        # Check pop until an existing timestamp
        pop_last_until_2 = data_list.pop_last_until(2)

        self.assertEqual(pop_last_until_2.message, "C")
        self.assertEqual(pop_last_until_2.timestamp, 2)
        self.assertEqual(len(data_list), 2)
        self.assertEqual(data_list[0].message, "C")
        self.assertEqual(data_list[0].timestamp, 2)

        # Check first until an none existing timestamp
        data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \
        [
            BasicDataClass(message="A", timestamp=0),
            BasicDataClass(message="B", timestamp=1),
            BasicDataClass(message="C", timestamp=2),
            BasicDataClass(message="D", timestamp=3)
        ])

        pop_last_until_1dot5 = data_list.pop_last_until(1.5)

        self.assertEqual(pop_last_until_1dot5.message, "B")
        self.assertEqual(pop_last_until_1dot5.timestamp, 1)
        self.assertEqual(len(data_list), 3)
        self.assertEqual(data_list[0].message, "B")
        self.assertEqual(data_list[0].timestamp, 1)

    def test_pop_last_before(self):
        """Test TimestampedObjectsList pop_last_before method."""

        data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \
        [
            BasicDataClass(message="A", timestamp=0),
            BasicDataClass(message="B", timestamp=1),
            BasicDataClass(message="C", timestamp=2),
            BasicDataClass(message="D", timestamp=3)
        ])

        # Check pop until an existing timestamp
        last_before_2 = data_list.pop_last_before(2)

        self.assertEqual(last_before_2.message, "B")
        self.assertEqual(last_before_2.timestamp, 1)
        self.assertEqual(len(data_list), 2)
        self.assertEqual(data_list[0].message, "C")
        self.assertEqual(data_list[0].timestamp, 2)

        # Check pop until an none existing timestamp
        data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \
        [
            BasicDataClass(message="A", timestamp=0),
            BasicDataClass(message="B", timestamp=1),
            BasicDataClass(message="C", timestamp=2),
            BasicDataClass(message="D", timestamp=3)
        ])

        first_until_1dot5 = data_list.pop_last_before(1.5)

        self.assertEqual(first_until_1dot5.message, "B")
        self.assertEqual(first_until_1dot5.timestamp, 1)
        self.assertEqual(len(data_list), 2)
        self.assertEqual(data_list[0].message, "C")
        self.assertEqual(data_list[0].timestamp, 2)

    def test_pop_last(self):
        """Test TimestampedObjectsList pop_last method."""

        data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \
            [
                BasicDataClass(message="A", timestamp=0),
                BasicDataClass(message="B", timestamp=1)
            ])

        last = data_list.pop(-1)

        self.assertEqual(len(data_list), 1)
        self.assertEqual(last.message, "B")
        self.assertEqual(last.timestamp, 1)
        self.assertEqual(data_list[0].message, "A")
        self.assertEqual(data_list[0].timestamp, 0)

    def test_get_first_from(self):
        """Test TimestampedObjectsList get_first_from method."""

        data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \
        [
            BasicDataClass(message="A", timestamp=0),
            BasicDataClass(message="B", timestamp=1),
            BasicDataClass(message="C", timestamp=2),
            BasicDataClass(message="D", timestamp=3)
        ])

        get_first_from_1 = data_list.get_first_from(1)

        self.assertEqual(get_first_from_1.message, "B")
        self.assertEqual(get_first_from_1.timestamp, 1)
        self.assertEqual(len(data_list), 4)

        get_first_from_1dot5 = data_list.get_first_from(1.5)

        self.assertEqual(get_first_from_1dot5.message, "C")
        self.assertEqual(get_first_from_1dot5.timestamp, 2)

        get_first_from_0 = data_list.get_first_from(0)

        self.assertEqual(get_first_from_0.message, "A")
        self.assertEqual(get_first_from_0.timestamp, 0)

        # Check that accessing to lately timestamp fails
        with self.assertRaises(KeyError):

            data_list.get_first_from(4)

    def test_get_last_before(self):
        """Test TimestampedObjectsList get_last_before method."""

        data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \
        [
            BasicDataClass(message="A", timestamp=0),
            BasicDataClass(message="B", timestamp=1),
            BasicDataClass(message="C", timestamp=2),
            BasicDataClass(message="D", timestamp=3)
        ])

        get_last_before_2 = data_list.get_last_before(2)

        self.assertEqual(get_last_before_2.message, "B")
        self.assertEqual(get_last_before_2.timestamp, 1)
        self.assertEqual(len(data_list), 4)

        get_last_before_1dot5 = data_list.get_last_before(1.5)

        self.assertEqual(get_last_before_1dot5.message,"B")
        self.assertEqual(get_last_before_1dot5.timestamp, 1)

        get_last_before_4 = data_list.get_last_before(4)

        self.assertEqual(get_last_before_4.message, "D")
        self.assertEqual(get_last_before_4.timestamp, 3)

        # Check that accessing to early timestamp fails
        with self.assertRaises(KeyError):

            data_list.get_last_before(-1)  
      
    def test_get_last_until(self):
        """Test TimestampedObjectsList get_last_until method."""

        data_list = DataFeatures.TimestampedObjectsList(BasicDataClass, \
        [
            BasicDataClass(message="A", timestamp=0),
            BasicDataClass(message="B", timestamp=1),
            BasicDataClass(message="C", timestamp=2),
            BasicDataClass(message="D", timestamp=3)
        ])

        get_last_until_2 = data_list.get_last_until(2)

        self.assertEqual(get_last_until_2.message, "C")
        self.assertEqual(get_last_until_2.timestamp, 2)
        self.assertEqual(len(data_list), 4)

        get_last_until_1dot5 = data_list.get_last_until(1.5)

        self.assertEqual(get_last_until_1dot5.message, "B")
        self.assertEqual(get_last_until_1dot5.timestamp, 1)

        get_last_until_4 = data_list.get_last_until(4)

        self.assertEqual(get_last_until_4.message, "D")
        self.assertEqual(get_last_until_4.timestamp, 3)

        # Check that accessing to early timestamp fails
        with self.assertRaises(KeyError):

            data_list.get_last_until(-1)

if __name__ == '__main__':

    unittest.main()