bart/common/signal.py - platform/external/bart - Git at Google

 #    Copyright 2015-2016 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.
 #

 """
 **Signals**

     - Definition

         A signal is a string representation of a TRAPpy event and the
         column in the same event. The signal can be of two types:

             - *Pivoted Signal*

                 A pivoted signal has a pivot specified in its event class.
                 This means that the signal in the event is a concatenation of different
                 signals which belong to different **pivot** nodes. The analysis for pivoted
                 signals must be done by decomposing them into pivoted signals for each node.

                 For example, an even that represents the load of the CPU can be pivoted on
                 :code:`"cpu"` which should be a column in the event's `DataFrame`

             - *Non-Pivoted Signal*

                 A non pivoted signal has an event that has no pivot value associated with it.
                 This probably means that signal has one component and can be analysed without
                 decomposing it into smaller signals.

     - Representation

         The following are valid representations of a signal

         - :code:`"event_name:event_column"`
         - :code:`"trappy.event.class:event_column"`

 """

 from trappy.stats.grammar import Parser
 from trappy.stats import StatConf
 from bart.common.Utils import area_under_curve, interval_sum

 # pylint: disable=invalid-name
 # pylint: disable=anomalous-backslash-in-string

 class SignalCompare(object):

     """
     :param data: TRAPpy FTrace Object
     :type data: :mod:`trappy.ftrace.FTrace`

     :param sig_a: The first signal
     :type sig_a: str

     :param sig_b: The first signal
     :type sig_b: str

     :param config: A dictionary of variables, classes
         and functions that can be used in the statements
     :type config: dict

     :param method: The method to be used for reindexing data
         This can be one of the standard :mod:`pandas.DataFrame`
         methods (eg. pad, bfill, nearest). The default is pad
         or use the last valid observation.
     :type method: str

     :param limit: The number of indices a value will be propagated
         when reindexing. The default is None
     :type limit: int

     :param fill: Whether to fill the NaNs in the data.
         The default value is True.
     :type fill: bool

     .. note::

         Both the signals must have the same pivots. For example:

             - Signal A has a pivot as :code:`"cpu"` which means that
               the trappy event (:mod:`trappy.base.Base`) has a pivot
               parameter which is equal to :code:`"cpu"`. Then the signal B
               should also have :code:`"cpu"` as it's pivot.

             - Signal A and B can both have undefined or None
               as their pivots
     """

     def __init__(self, data, sig_a, sig_b, **kwargs):

         self._parser = Parser(
             data,
             config=kwargs.pop(
                 "config",
                 None),
             **kwargs)
         self._a = sig_a
         self._b = sig_b
         self._pivot_vals, self._pivot = self._get_signal_pivots()

         # Concatenate the indices by doing any operation (say add)
         self._a_data = self._parser.solve(sig_a)
         self._b_data = self._parser.solve(sig_b)

     def _get_signal_pivots(self):
         """Internal function to check pivot conditions and
         return an intersection of pivot on the signals"""

         sig_a_info = self._parser.inspect(self._a)
         sig_b_info = self._parser.inspect(self._b)

         if sig_a_info["pivot"] != sig_b_info["pivot"]:
             raise RuntimeError("The pivot column for both signals" +
                                "should be same (%s,%s)"
                                % (sig_a_info["pivot"], sig_b_info["pivot"]))

         if sig_a_info["pivot"]:
             pivot_vals = set(
                 sig_a_info["pivot_values"]).intersection(sig_b_info["pivot_values"])
             pivoted = sig_a_info["pivot"]
         else:
             pivot_vals = [StatConf.GRAMMAR_DEFAULT_PIVOT]
             pivoted = False

         return pivot_vals, pivoted

     def conditional_compare(self, condition, **kwargs):
         """Conditionally compare two signals

         The conditional comparison of signals has two components:

         - **Value Coefficient** :math:`\\alpha_{v}` which measures the difference in values of
           of the two signals when the condition is true:

           .. math::

                 \\alpha_{v} = \\frac{area\_under\_curve(S_A\ |\ C(t)\ is\ true)}
                 {area\_under\_curve(S_B\ |\ C(t)\ is\ true)} \\\\

                 \\alpha_{v} = \\frac{\int S_A(\{t\ |\ C(t)\})dt}{\int S_B(\{t\ |\ C(t)\})dt}

         - **Time Coefficient** :math:`\\alpha_{t}` which measures the time during which the
           condition holds true.

           .. math::

                 \\alpha_{t} = \\frac{T_{valid}}{T_{total}}

         :param condition: A condition that returns a truth value and obeys the grammar syntax
             ::

                 "event_x:sig_a > event_x:sig_b"

         :type condition: str

         :param method: The method for area calculation. This can
             be any of the integration methods supported in `numpy`
             or `rect`
         :type param: str

         :param step: The step behaviour for area and time
             summation calculation
         :type step: str

         Consider the two signals A and B as follows:

             .. code::

                 A = [0, 0, 0, 3, 3, 0, 0, 0]
                 B = [0, 0, 2, 2, 2, 2, 1, 1]


             .. code::


                                                      A = xxxx
                 3                 *xxxx*xxxx+        B = ----
                                   |         |
                 2            *----*----*----+
                              |    |         |
                 1            |    |         *----*----+
                              |    |         |
                 0  *x-x-*x-x-+xxxx+         +xxxx*xxxx+
                    0    1    2    3    4    5    6    7

         The condition:

         .. math::

             A > B

         is valid between T=3 and T=5. Therefore,

         .. math::

             \\alpha_v=1.5 \\\\
             \\alpha_t=\\frac{2}{7}

         :returns: There are two cases:

             - **Pivoted Signals**
               ::

                     {
                         "pivot_name" : {
                                 "pval_1" : (v1,t1),
                                 "pval_2" : (v2, t2)
                         }
                     }
             - **Non Pivoted Signals**

               The tuple of :math:`(\\alpha_v, \\alpha_t)`
         """

         if self._pivot:
             result = {self._pivot: {}}

         mask = self._parser.solve(condition)
         step = kwargs.get("step", "post")

         for pivot_val in self._pivot_vals:

             a_piv = self._a_data[pivot_val]
             b_piv = self._b_data[pivot_val]

             area = area_under_curve(a_piv[mask[pivot_val]], **kwargs)
             try:
                 area /= area_under_curve(b_piv[mask[pivot_val]], **kwargs)
             except ZeroDivisionError:
                 area = float("nan")

             duration = min(a_piv.last_valid_index(), b_piv.last_valid_index())
             duration -= max(a_piv.first_valid_index(),
                             b_piv.first_valid_index())
             duration = interval_sum(mask[pivot_val], step=step) / duration

             if self._pivot:
                 result[self._pivot][pivot_val] = area, duration
             else:
                 result = area, duration

         return result

     def get_overshoot(self, **kwargs):
         """Special case for :func:`conditional_compare`
         where the condition is:
         ::

             "sig_a > sig_b"

         :param method: The method for area calculation. This can
             be any of the integration methods supported in `numpy`
             or `rect`
         :type param: str

         :param step: The step behaviour for calculation of area
             and time summation
         :type step: str

         .. seealso::

             :func:`conditional_compare`
         """

         condition = " ".join([self._a, ">", self._b])
         return self.conditional_compare(condition, **kwargs)

     def get_undershoot(self, **kwargs):
         """Special case for :func:`conditional_compare`
         where the condition is:
         ::

             "sig_a < sig_b"

         :param method: The method for area calculation. This can
             be any of the integration methods supported in `numpy`
             or `rect`
         :type param: str

         :param step: The step behaviour for calculation of area
             and time summation
         :type step: str

         .. seealso::

             :func:`conditional_compare`
         """

         condition = " ".join([self._a, "<", self._b])
         return self.conditional_compare(condition, **kwargs)
	# Copyright 2015-2016 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.
	#

	"""
	Signals

	- Definition

	A signal is a string representation of a TRAPpy event and the
	column in the same event. The signal can be of two types:

	- Pivoted Signal

	A pivoted signal has a pivot specified in its event class.
	This means that the signal in the event is a concatenation of different
	signals which belong to different pivot nodes. The analysis for pivoted
	signals must be done by decomposing them into pivoted signals for each node.

	For example, an even that represents the load of the CPU can be pivoted on
	:code:`"cpu"` which should be a column in the event's `DataFrame`

	- Non-Pivoted Signal

	A non pivoted signal has an event that has no pivot value associated with it.
	This probably means that signal has one component and can be analysed without
	decomposing it into smaller signals.

	- Representation

	The following are valid representations of a signal

	- :code:`"event_name:event_column"`
	- :code:`"trappy.event.class:event_column"`

	"""

	from trappy.stats.grammar import Parser
	from trappy.stats import StatConf
	from bart.common.Utils import area_under_curve, interval_sum

	# pylint: disable=invalid-name
	# pylint: disable=anomalous-backslash-in-string

	class SignalCompare(object):

	"""
	:param data: TRAPpy FTrace Object
	:type data: :mod:`trappy.ftrace.FTrace`

	:param sig_a: The first signal
	:type sig_a: str

	:param sig_b: The first signal
	:type sig_b: str

	:param config: A dictionary of variables, classes
	and functions that can be used in the statements
	:type config: dict

	:param method: The method to be used for reindexing data
	This can be one of the standard :mod:`pandas.DataFrame`
	methods (eg. pad, bfill, nearest). The default is pad
	or use the last valid observation.
	:type method: str

	:param limit: The number of indices a value will be propagated
	when reindexing. The default is None
	:type limit: int

	:param fill: Whether to fill the NaNs in the data.
	The default value is True.
	:type fill: bool

	.. note::

	Both the signals must have the same pivots. For example:

	- Signal A has a pivot as :code:`"cpu"` which means that
	the trappy event (:mod:`trappy.base.Base`) has a pivot
	parameter which is equal to :code:`"cpu"`. Then the signal B
	should also have :code:`"cpu"` as it's pivot.

	- Signal A and B can both have undefined or None
	as their pivots
	"""

	def __init__(self, data, sig_a, sig_b, **kwargs):

	self._parser = Parser(
	data,
	config=kwargs.pop(
	"config",
	None),
	**kwargs)
	self._a = sig_a
	self._b = sig_b
	self._pivot_vals, self._pivot = self._get_signal_pivots()

	# Concatenate the indices by doing any operation (say add)
	self._a_data = self._parser.solve(sig_a)
	self._b_data = self._parser.solve(sig_b)

	def _get_signal_pivots(self):
	"""Internal function to check pivot conditions and
	return an intersection of pivot on the signals"""

	sig_a_info = self._parser.inspect(self._a)
	sig_b_info = self._parser.inspect(self._b)

	if sig_a_info["pivot"] != sig_b_info["pivot"]:
	raise RuntimeError("The pivot column for both signals" +
	"should be same (%s,%s)"
	% (sig_a_info["pivot"], sig_b_info["pivot"]))

	if sig_a_info["pivot"]:
	pivot_vals = set(
	sig_a_info["pivot_values"]).intersection(sig_b_info["pivot_values"])
	pivoted = sig_a_info["pivot"]
	else:
	pivot_vals = [StatConf.GRAMMAR_DEFAULT_PIVOT]
	pivoted = False

	return pivot_vals, pivoted

	def conditional_compare(self, condition, **kwargs):
	"""Conditionally compare two signals

	The conditional comparison of signals has two components:

	- Value Coefficient :math:`\\alpha_{v}` which measures the difference in values of
	of the two signals when the condition is true:

	.. math::

	\\alpha_{v} = \\frac{area\_under\_curve(S_A\ \|\ C(t)\ is\ true)}
	{area\_under\_curve(S_B\ \|\ C(t)\ is\ true)} \\\\

	\\alpha_{v} = \\frac{\int S_A(\{t\ \|\ C(t)\})dt}{\int S_B(\{t\ \|\ C(t)\})dt}

	- Time Coefficient :math:`\\alpha_{t}` which measures the time during which the
	condition holds true.

	.. math::

	\\alpha_{t} = \\frac{T_{valid}}{T_{total}}

	:param condition: A condition that returns a truth value and obeys the grammar syntax
	::

	"event_x:sig_a > event_x:sig_b"

	:type condition: str

	:param method: The method for area calculation. This can
	be any of the integration methods supported in `numpy`
	or `rect`
	:type param: str

	:param step: The step behaviour for area and time
	summation calculation
	:type step: str

	Consider the two signals A and B as follows:

	.. code::

	A = [0, 0, 0, 3, 3, 0, 0, 0]
	B = [0, 0, 2, 2, 2, 2, 1, 1]


	.. code::


	A = xxxx
	3 xxxxxxxx+ B = ----
	\| \|
	2 --------*----+
	\| \| \|
	1 \| \| --------+
	\| \| \|
	0 x-x-x-x-+xxxx+ +xxxx*xxxx+
	0 1 2 3 4 5 6 7

	The condition:

	.. math::

	A > B

	is valid between T=3 and T=5. Therefore,

	.. math::

	\\alpha_v=1.5 \\\\
	\\alpha_t=\\frac{2}{7}

	:returns: There are two cases:

	- Pivoted Signals
	::

	{
	"pivot_name" : {
	"pval_1" : (v1,t1),
	"pval_2" : (v2, t2)
	}
	}
	- Non Pivoted Signals

	The tuple of :math:`(\\alpha_v, \\alpha_t)`
	"""

	if self._pivot:
	result = {self._pivot: {}}

	mask = self._parser.solve(condition)
	step = kwargs.get("step", "post")

	for pivot_val in self._pivot_vals:

	a_piv = self._a_data[pivot_val]
	b_piv = self._b_data[pivot_val]

	area = area_under_curve(a_piv[mask[pivot_val]], **kwargs)
	try:
	area /= area_under_curve(b_piv[mask[pivot_val]], **kwargs)
	except ZeroDivisionError:
	area = float("nan")

	duration = min(a_piv.last_valid_index(), b_piv.last_valid_index())
	duration -= max(a_piv.first_valid_index(),
	b_piv.first_valid_index())
	duration = interval_sum(mask[pivot_val], step=step) / duration

	if self._pivot:
	result[self._pivot][pivot_val] = area, duration
	else:
	result = area, duration

	return result

	def get_overshoot(self, **kwargs):
	"""Special case for :func:`conditional_compare`
	where the condition is:
	::

	"sig_a > sig_b"

	:param method: The method for area calculation. This can
	be any of the integration methods supported in `numpy`
	or `rect`
	:type param: str

	:param step: The step behaviour for calculation of area
	and time summation
	:type step: str

	.. seealso::

	:func:`conditional_compare`
	"""

	condition = " ".join([self._a, ">", self._b])
	return self.conditional_compare(condition, **kwargs)

	def get_undershoot(self, **kwargs):
	"""Special case for :func:`conditional_compare`
	where the condition is:
	::

	"sig_a < sig_b"

	:param method: The method for area calculation. This can
	be any of the integration methods supported in `numpy`
	or `rect`
	:type param: str

	:param step: The step behaviour for calculation of area
	and time summation
	:type step: str

	.. seealso::

	:func:`conditional_compare`
	"""

	condition = " ".join([self._a, "<", self._b])
	return self.conditional_compare(condition, **kwargs)