tests/test_stats.py

#    Copyright 2015-2017 ARM Limited
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

import unittest
from trappy.stats.Topology import Topology
from trappy.stats.Trigger import Trigger
from trappy.stats.Aggregator import MultiTriggerAggregator

import collections
import trappy
from trappy.base import Base
import pandas as pd
from pandas.util.testing import assert_series_equal


class TestTopology(unittest.TestCase):

    def test_add_to_level(self):
        """Test level creation"""

        level_groups = [[1, 2], [0, 3, 4, 5]]
        level = "test_level"
        topology = Topology()
        topology.add_to_level(level, level_groups)
        check_groups = topology.get_level(level)

        self.assertTrue(topology.has_level(level))
        self.assertEqual(level_groups, check_groups)

    def test_flatten(self):
        """Test Topology: flatten"""

        level_groups = [[1, 2], [0, 3, 4, 5]]
        level = "test_level"
        topology = Topology()
        topology.add_to_level(level, level_groups)
        flattened = [0, 1, 2, 3, 4, 5]

        self.assertEqual(flattened, topology.flatten())

    def test_cpu_topology_construction(self):
        """Test CPU Topology Construction"""

        cluster_0 = [0, 3, 4, 5]
        cluster_1 = [1, 2]
        clusters = [cluster_0, cluster_1]
        topology = Topology(clusters=clusters)

        # Check cluster level creation
        cluster_groups = [[0, 3, 4, 5], [1, 2]]
        self.assertTrue(topology.has_level("cluster"))
        self.assertEqual(cluster_groups, topology.get_level("cluster"))

        # Check cpu level creation
        cpu_groups = [[0], [1], [2], [3], [4], [5]]
        self.assertTrue(topology.has_level("cpu"))
        self.assertEqual(cpu_groups, topology.get_level("cpu"))

        # Check "all" level
        all_groups = [[0, 1, 2, 3, 4, 5]]
        self.assertEqual(all_groups, topology.get_level("all"))

    def test_level_span(self):
        """TestTopology: level_span"""

        level_groups = [[1, 2], [0, 3, 4, 5]]
        level = "test_level"
        topology = Topology()
        topology.add_to_level(level, level_groups)

        self.assertEqual(topology.level_span(level), 2)

    def test_group_index(self):
        """TestTopology: get_index"""

        level_groups = [[1, 2], [0, 3, 4, 5]]
        level = "test_level"
        topology = Topology()
        topology.add_to_level(level, level_groups)

        self.assertEqual(topology.get_index(level, [1, 2]), 0)
        self.assertEqual(topology.get_index(level, [0, 3, 4, 5]), 1)

class BaseTestStats(unittest.TestCase):
    def setUp(self):
        trace = trappy.BareTrace()
        data = {

            "identifier": [
                0,
                0,
                0,
                1,
                1,
                1,
            ],
            "result": [
                "fire",
                "blank",
                "fire",
                "blank",
                "fire",
                "blank",
            ],
        }

        index = pd.Series([0.1, 0.2, 0.3, 0.4, 0.5, 0.6], name="Time")
        data_frame = pd.DataFrame(data, index=index)
        trace.add_parsed_event("aim_and_fire", data_frame)
        self._trace = trace
        self.topology = Topology(clusters=[[0], [1]])


class TestTrigger(BaseTestStats):

    def test_trigger_generation(self):
        """TestTrigger: generate"""

        filters = {
            "result": "fire"
        }

        event_class = self._trace.aim_and_fire
        value = 1
        pivot = "identifier"

        trigger = Trigger(self._trace,
                          event_class,
                          filters,
                          value,
                          pivot)

        expected = pd.Series([1, 1], index=pd.Index([0.1, 0.3], name="Time"))
        assert_series_equal(expected, trigger.generate(0))

        expected = pd.Series([1], index=pd.Index([0.5], name="Time"))
        assert_series_equal(expected, trigger.generate(1))

    def test_trigger_with_func(self):
        """Trigger works with a function or lambda as filter"""

        def my_filter(val):
            return val.startswith("fi")

        trigger = Trigger(self._trace, self._trace.aim_and_fire,
                          filters={"result": my_filter}, value=1,
                          pivot="identifier")

        expected = pd.Series([1], index=pd.Index([0.5], name="Time"))
        assert_series_equal(expected, trigger.generate(1))

        my_filters = {"result": lambda x: x.startswith("bl")}
        trigger = Trigger(self._trace, self._trace.aim_and_fire,
                          filters=my_filters, value=1, pivot="identifier")

        expected = pd.Series([1, 1], index=pd.Index([0.4, 0.6], name="Time"))
        assert_series_equal(expected, trigger.generate(1))

    def test_trigger_with_callable_class(self):
        """Trigger works with a callable class as filter"""

        class my_filter(object):
            def __init__(self, val_out):
                self.prev_val = 0
                self.val_out = val_out

            def __call__(self, val):
                ret = self.prev_val == self.val_out
                self.prev_val = val

                return ret

        trigger = Trigger(self._trace, self._trace.aim_and_fire,
                          filters={"identifier": my_filter(1)}, value=1,
                          pivot="result")

        expected = pd.Series([1], index=pd.Index([0.6], name="Time"))
        assert_series_equal(expected, trigger.generate("blank"))

    def test_filter_prev_values(self):
        """Trigger works with a filter that depends on previous values of the same pivot"""

        # We generate an example in which we want a trigger whenever the
        # identifier is no longer 1 for blank

        class my_filter(object):
            def __init__(self, val_out):
                self.prev_val = 0
                self.val_out = val_out

            def __call__(self, val):
                ret = self.prev_val == self.val_out
                self.prev_val = val

                return ret

        trace = trappy.BareTrace()
        data = collections.OrderedDict([
            (0.1, ["blank", 1]),
            (0.2, ["fire",  1]),
            (0.3, ["blank", 0]), # value is no longer 1, trigger
            (0.4, ["blank", 1]),
            (0.5, ["fire",  0]), # This should NOT trigger
            (0.6, ["blank", 0]), # value is no longer 1 for blank, trigger
        ])
        data_frame = pd.DataFrame.from_dict(data, orient="index", )
        data_frame.columns = ["result", "identifier"]
        trace.add_parsed_event("aim_and_fire", data_frame)

        trigger = Trigger(trace, trace.aim_and_fire,
                          filters={"identifier": my_filter(1)}, value=-1,
                          pivot="result")

        expected = pd.Series([-1, -1], index=[0.3, 0.6])
        assert_series_equal(expected, trigger.generate("blank"))



class TestAggregator(BaseTestStats):

    def test_scalar_aggfunc_single_trigger(self):
        """TestAggregator: 1 trigger scalar aggfunc"""

        def aggfunc(series):
            return series.sum()

        filters = {
            "result": "fire"
        }

        event_class = self._trace.aim_and_fire
        value = 1
        pivot = "identifier"

        trigger = Trigger(self._trace,
                          event_class,
                          filters,
                          value,
                          pivot)

        aggregator = MultiTriggerAggregator([trigger],
                        self.topology,
                        aggfunc=aggfunc)

        # There are three "fire" in total
        # The all level in topology looks like
        # [[0, 1]]
        result = aggregator.aggregate(level="all")
        self.assertEqual(result, [3.0])

        # There are two "fire" on the first node group and a
        # a single "fire" on the second node group at the cluster
        # level which looks like
        # [[0], [1]]
        result = aggregator.aggregate(level="cluster")
        self.assertEqual(result, [2.0, 1.0])

    def test_vector_aggfunc_single_trigger(self):
        """TestAggregator: 1 trigger vector aggfunc"""

        def aggfunc(series):
            return series.cumsum()

        filters = {
            "result": "fire"
        }

        event_class = self._trace.aim_and_fire
        value = 1
        pivot = "identifier"

        trigger = Trigger(self._trace, event_class, filters, value, pivot)

        aggregator = MultiTriggerAggregator([trigger],
                        self.topology,
                        aggfunc=aggfunc)

        # There are three "fire" in total
        # The all level in topology looks like
        # [[0, 1]]
        result = aggregator.aggregate(level="all")
        expected_result = pd.Series([1.0, 1.0, 2.0, 2.0, 3.0, 3.0],
                                      index=pd.Index([0.1, 0.2, 0.3, 0.4, 0.5, 0.6])
                          )
        assert_series_equal(result[0], expected_result)

    def test_vector_aggfunc_multiple_trigger(self):
        """TestAggregator: multi trigger vector aggfunc"""

        def aggfunc(series):
            return series.cumsum()

        filters = {
            "result": "fire"
        }

        event_class = self._trace.aim_and_fire
        value = 1
        pivot = "identifier"

        trigger_fire = Trigger(self._trace,
                          event_class,
                          filters,
                          value,
                          pivot)

        filters = {
            "result": "blank"
        }
        value = -1
        trigger_blank = Trigger(self._trace, event_class, filters, value,
                                pivot)

        aggregator = MultiTriggerAggregator([trigger_fire, trigger_blank],
                        self.topology,
                        aggfunc=aggfunc)

        # There are three "fire" in total
        # The all level in topology looks like
        # [[0, 1]]
        result = aggregator.aggregate(level="all")
        expected_result = pd.Series([1.0, 0.0, 1.0, 0.0, 1.0, 0.0],
                                      index=pd.Index([0.1, 0.2, 0.3, 0.4, 0.5, 0.6])
                          )
        assert_series_equal(result[0], expected_result)

    def test_default_aggfunc_multiple_trigger(self):
        """MultiTriggerAggregator with the default aggfunc"""

        trigger_fire = Trigger(self._trace, self._trace.aim_and_fire,
                          filters={"result": "fire"},
                          pivot="identifier", value=1)

        trigger_blank = Trigger(self._trace, self._trace.aim_and_fire,
                          filters={"result": "blank"},
                          pivot="identifier", value=2)

        aggregator = MultiTriggerAggregator([trigger_fire, trigger_blank],
                                            self.topology)

        results = aggregator.aggregate(level="cpu")
        expected_results = [
            pd.Series([1., 2., 1., 0., 0., 0.],
                      index=[0.1, 0.2, 0.3, 0.4, 0.5, 0.6]),
            pd.Series([0., 0., 0., 2., 1., 2.],
                      index=[0.1, 0.2, 0.3, 0.4, 0.5, 0.6]),
            ]

        self.assertEquals(len(results), len(expected_results))
        for result, expected in zip(results, expected_results):
            assert_series_equal(result, expected)