path: root/libs/utils/trace.py

# SPDX-License-Identifier: Apache-2.0
#
# Copyright (C) 2015, ARM Limited and contributors.
#
# 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.
#

""" Trace Parser Module """

import numpy as np
import os
import pandas as pd
import sys
import trappy
import json
import warnings
import operator
import logging

from analysis_register import AnalysisRegister
from collections import namedtuple
from devlib.utils.misc import memoized
from trappy.utils import listify, handle_duplicate_index


NON_IDLE_STATE = -1
ResidencyTime = namedtuple('ResidencyTime', ['total', 'active'])
ResidencyData = namedtuple('ResidencyData', ['label', 'residency'])

class Trace(object):
    """
    The Trace object is the LISA trace events parser.

    :param platform: a dictionary containing information about the target
        platform
    :type platform: dict

    :param data_dir: folder containing all trace data
    :type data_dir: str

    :param events: events to be parsed (everything in the trace by default)
    :type events: list(str)

    :param tasks: filter data for the specified tasks only. If None (default),
        use data for all tasks found in the trace.
    :type tasks: list(str) or NoneType

    :param window: time window to consider when parsing the trace
    :type window: tuple(int, int)

    :param normalize_time: normalize trace time stamps
    :type normalize_time: bool

    :param trace_format: format of the trace. Possible values are:
        - FTrace
        - SysTrace
    :type trace_format: str

    :param plots_dir: directory where to save plots
    :type plots_dir: str

    :param plots_prefix: prefix for plots file names
    :type plots_prefix: str

    :param cgroup_info: add cgroup information for sanitization
      example:
        {
            'controller_ids': { 2: 'schedtune', 4: 'cpuset' },
            'cgroups': [ 'root', 'background', 'foreground' ],  # list of allowed cgroup names
        }
    :type cgroup_info: dict
    """

    def __init__(self, platform, data_dir, events=None,
                 tasks=None, window=(0, None),
                 normalize_time=True,
                 trace_format='FTrace',
                 plots_dir=None,
                 plots_prefix='',
                 cgroup_info={}):

        # The platform used to run the experiments
        self.platform = platform

        # TRAPpy Trace object
        self.ftrace = None

        # Trace format
        self.trace_format = trace_format

        # The time window used to limit trace parsing to
        self.window = window

        # Dynamically registered TRAPpy events
        self.trappy_cls = {}

        # Maximum timespan for all collected events
        self.time_range = 0

        # Time the system was overutilzied
        self.overutilized_time = 0
        self.overutilized_prc = 0

        # The dictionary of tasks descriptors available in the dataset
        self.tasks = {}

        # List of events required by user
        self.events = []

        # List of events available in the parsed trace
        self.available_events = []

        # Cluster frequency coherency flag
        self.freq_coherency = True

        # Folder containing all trace data
        self.data_dir = None

        # Setup logging
        self._log = logging.getLogger('Trace')

        # Folder containing trace
        if not os.path.isdir(data_dir):
            self.data_dir = os.path.dirname(data_dir)
        else:
            self.data_dir = data_dir

        # By deafult, use the trace dir to save plots
        self.plots_dir = plots_dir
        if self.plots_dir is None:
            self.plots_dir = self.data_dir
        self.plots_prefix = plots_prefix

        # Cgroup info for sanitization
        self.cgroup_info = cgroup_info

        self.__registerTraceEvents(events) if events else None
        self.__parseTrace(data_dir, tasks, window, normalize_time,
                          trace_format)
        self.__computeTimeSpan()

        # Minimum and Maximum x_time to use for all plots
        self.x_min = 0
        self.x_max = self.time_range

        # Reset x axis time range to full scale
        t_min = self.window[0]
        t_max = self.window[1]
        self.setXTimeRange(t_min, t_max)

        self.data_frame = TraceData()
        self._registerDataFrameGetters(self)

        self.analysis = AnalysisRegister(self)

    def _registerDataFrameGetters(self, module):
        """
        Internal utility function that looks up getter functions with a "_dfg_"
        prefix in their name and bounds them to the specified module.

        :param module: module to which the function is added
        :type module: class
        """
        self._log.debug('Registering [%s] local data frames', module)
        for func in dir(module):
            if not func.startswith('_dfg_'):
                continue
            dfg_name = func.replace('_dfg_', '')
            dfg_func = getattr(module, func)
            self._log.debug('   %s', dfg_name)
            setattr(self.data_frame, dfg_name, dfg_func)

    def setXTimeRange(self, t_min=None, t_max=None):
        """
        Set x axis time range to the specified values.

        :param t_min: lower bound
        :type t_min: int or float

        :param t_max: upper bound
        :type t_max: int or float
        """
        if t_min is None:
            self.x_min = 0
        else:
            self.x_min = t_min
        if t_max is None:
            self.x_max = self.time_range
        else:
            self.x_max = t_max
        self._log.debug('Set plots time range to (%.6f, %.6f)[s]',
                       self.x_min, self.x_max)

    def __registerTraceEvents(self, events):
        """
        Save a copy of the parsed events.

        :param events: single event name or list of events names
        :type events: str or list(str)
        """
        if isinstance(events, basestring):
            self.events = events.split(' ')
        elif isinstance(events, list):
            self.events = events
        else:
            raise ValueError('Events must be a string or a list of strings')
        # Register devlib fake cpu_frequency events
        if 'cpu_frequency' in events:
            self.events.append('cpu_frequency_devlib')

    def __parseTrace(self, path, tasks, window, normalize_time, trace_format):
        """
        Internal method in charge of performing the actual parsing of the
        trace.

        :param path: path to the trace folder (or trace file)
        :type path: str

        :param tasks: filter data for the specified tasks only
        :type tasks: list(str)

        :param window: time window to consider when parsing the trace
        :type window: tuple(int, int)

        :param normalize_time: normalize trace time stamps
        :type normalize_time: bool

        :param trace_format: format of the trace. Possible values are:
            - FTrace
            - SysTrace
        :type trace_format: str
        """
        self._log.debug('Loading [sched] events from trace in [%s]...', path)
        self._log.debug('Parsing events: %s', self.events if self.events else 'ALL')
        if trace_format.upper() == 'SYSTRACE' or path.endswith('html'):
            self._log.debug('Parsing SysTrace format...')
            trace_class = trappy.SysTrace
            self.trace_format = 'SysTrace'
        elif trace_format.upper() == 'FTRACE':
            self._log.debug('Parsing FTrace format...')
            trace_class = trappy.FTrace
            self.trace_format = 'FTrace'
        else:
            raise ValueError("Unknown trace format {}".format(trace_format))

        scope = 'custom' if self.events else 'all'
        self.ftrace = trace_class(path, scope=scope, events=self.events,
                                  window=window, normalize_time=normalize_time)

        # Load Functions profiling data
        has_function_stats = self._loadFunctionsStats(path)

        # Check for events available on the parsed trace
        self.__checkAvailableEvents()
        if len(self.available_events) == 0:
            if has_function_stats:
                self._log.info('Trace contains only functions stats')
                return
            raise ValueError('The trace does not contain useful events '
                             'nor function stats')

        # Sanitize cgroup info if any
        self._sanitize_CgroupAttachTask()

        # Santization not possible if platform missing
        if not self.platform:
            # Setup internal data reference to interesting events/dataframes
            self._sanitize_SchedLoadAvgCpu()
            self._sanitize_SchedLoadAvgTask()
            self._sanitize_SchedCpuCapacity()
            self._sanitize_SchedBoostCpu()
            self._sanitize_SchedBoostTask()
            self._sanitize_SchedEnergyDiff()
            self._sanitize_SchedOverutilized()
            self._sanitize_CpuFrequency()

        self.__loadTasksNames(tasks)

        # Compute plot window
        if not normalize_time:
            start = self.window[0]
            if self.window[1]:
                duration = min(self.ftrace.get_duration(), self.window[1])
            else:
                duration = self.ftrace.get_duration()
            self.window = (self.ftrace.basetime + start,
                           self.ftrace.basetime + duration)

    def __checkAvailableEvents(self, key=""):
        """
        Internal method used to build a list of available events.

        :param key: key to be used for TRAPpy filtering
        :type key: str
        """
        for val in self.ftrace.get_filters(key):
            obj = getattr(self.ftrace, val)
            if len(obj.data_frame):
                self.available_events.append(val)
        self._log.debug('Events found on trace:')
        for evt in self.available_events:
            self._log.debug(' - %s', evt)

    def __loadTasksNames(self, tasks):
        """
        Try to load tasks names using one of the supported events.

        :param tasks: list of task names. If None, load all tasks found.
        :type tasks: list(str) or NoneType
        """
        def load(tasks, event, name_key, pid_key):
            df = self._dfg_trace_event(event)
            if tasks is None:
                tasks = df[name_key].unique()
            self.getTasks(df, tasks, name_key=name_key, pid_key=pid_key)
            self._scanTasks(df, name_key=name_key, pid_key=pid_key)
            self._scanTgids(df)

        if 'sched_switch' in self.available_events:
            load(tasks, 'sched_switch', 'next_comm', 'next_pid')
        elif 'sched_load_avg_task' in self.available_events:
            load(tasks, 'sched_load_avg_task', 'comm', 'pid')
        else:
            self._log.warning('Failed to load tasks names from trace events')

    def hasEvents(self, dataset):
        """
        Returns True if the specified event is present in the parsed trace,
        False otherwise.

        :param dataset: trace event name or list of trace events
        :type dataset: str or list(str)
        """
        if dataset in self.available_events:
            return True
        return False

    def __computeTimeSpan(self):
        """
        Compute time axis range, considering all the parsed events.
        """
        ts = sys.maxint
        te = 0

        for events in self.available_events:
            df = self._dfg_trace_event(events)
            if len(df) == 0:
                continue
            if (df.index[0]) < ts:
                ts = df.index[0]
            if (df.index[-1]) > te:
                te = df.index[-1]
            self.time_range = te - ts

        self._log.debug('Collected events spans a %.3f [s] time interval',
                       self.time_range)

        # Build a stat on trace overutilization
        if self.hasEvents('sched_overutilized'):
            df = self._dfg_trace_event('sched_overutilized')
            self.overutilized_time = df[df.overutilized == 1].len.sum()
            self.overutilized_prc = 100. * self.overutilized_time / self.time_range

            self._log.debug('Overutilized time: %.6f [s] (%.3f%% of trace time)',
                           self.overutilized_time, self.overutilized_prc)

    def _scanTgids(self, df):
        if not '__tgid' in df.columns:
            return
        df = df[['__pid', '__tgid']]
        df = df.drop_duplicates(keep='first').set_index('__pid')
        df.rename(columns = { '__pid': 'pid', '__tgid': 'tgid' },
                              inplace=True)
        self._pid_tgid = df

    def _scanTasks(self, df, name_key='comm', pid_key='pid'):
        """
        Extract tasks names and PIDs from the input data frame. The data frame
        should contain a task name column and PID column.

        :param df: data frame containing trace events from which tasks names
            and PIDs will be extracted
        :type df: :mod:`pandas.DataFrame`

        :param name_key: The name of the dataframe columns containing task
            names
        :type name_key: str

        :param pid_key: The name of the dataframe columns containing task PIDs
        :type pid_key: str
        """
        df = df[[name_key, pid_key]].drop_duplicates()
        self._tasks_by_name = df.set_index(name_key)
        self._tasks_by_pid = df.set_index(pid_key)

    def getTaskByName(self, name):
        """
        Get the PIDs of all tasks with the specified name.

        :param name: task name
        :type name: str
        """
        if name not in self._tasks_by_name.index:
            return []
        if len(self._tasks_by_name.ix[name].values) > 1:
            return list({task[0] for task in
                         self._tasks_by_name.ix[name].values})
        return [self._tasks_by_name.ix[name].values[0]]

    def getTaskByPid(self, pid):
        """
        Get the names of all tasks with the specified PID.

        :param name: task PID
        :type name: int
        """
        if pid not in self._tasks_by_pid.index:
            return []
        if len(self._tasks_by_pid.ix[pid].values) > 1:
            return list({task[0] for task in
                         self._tasks_by_pid.ix[pid].values})
        return [self._tasks_by_pid.ix[pid].values[0]]

    def getTgidFromPid(self, pid):
        return _pid_tgid.ix[pid].values[0]

    def getTasks(self, dataframe=None,
                 task_names=None, name_key='comm', pid_key='pid'):
        """
        Helper function to get PIDs of specified tasks.

        This method can take a Pandas dataset in input to be used to fiter out
        the PIDs of all the specified tasks. If a dataset is not provided,
        previously filtered PIDs are returned.

        If a list of task names is not provided, all tasks detected in the trace
        will be used. The specified dataframe must provide at least two columns
        reporting the task name and the task PID. The default values of this
        colums could be specified using the provided parameters.

        :param dataframe: A Pandas dataframe containing at least 'name_key' and
            'pid_key' columns. If None, the all PIDs are returned.
        :type dataframe: :mod:`pandas.DataFrame`

        :param task_names: The list of tasks to get the PID of (default: all
            tasks)
        :type task_names: list(str)

        :param name_key: The name of the dataframe columns containing task
            names
        :type name_key: str

        :param pid_key: The name of the dataframe columns containing task PIDs
        :type pid_key: str
        """
        if task_names is None:
            task_names = self.tasks.keys()
        if dataframe is None:
            return {k: v for k, v in  self.tasks.iteritems() if k in task_names}
        df = dataframe
        self._log.debug('Lookup dataset for tasks...')
        for tname in task_names:
            self._log.debug('Lookup for task [%s]...', tname)
            results = df[df[name_key] == tname][[name_key, pid_key]]
            if len(results) == 0:
                self._log.error('  task %16s NOT found', tname)
                continue
            (name, pid) = results.head(1).values[0]
            if name != tname:
                self._log.error('  task %16s NOT found', tname)
                continue
            if tname not in self.tasks:
                self.tasks[tname] = {}
            pids = list(results[pid_key].unique())
            self.tasks[tname]['pid'] = pids
            self._log.debug('  task %16s found, pid: %s',
                            tname, self.tasks[tname]['pid'])
        return self.tasks


###############################################################################
# DataFrame Getter Methods
###############################################################################

    def df(self, event):
        """
        Get a dataframe containing all occurrences of the specified trace event
        in the parsed trace.

        :param event: Trace event name
        :type event: str
        """
        warnings.simplefilter('always', DeprecationWarning) #turn off filter
        warnings.warn("\n\tUse of Trace::df() is deprecated and will be soon removed."
                      "\n\tUse Trace::data_frame.trace_event(event_name) instead.",
                      category=DeprecationWarning)
        warnings.simplefilter('default', DeprecationWarning) #reset filter
        return self._dfg_trace_event(event)

    def _dfg_trace_event(self, event):
        """
        Get a dataframe containing all occurrences of the specified trace event
        in the parsed trace.

        :param event: Trace event name
        :type event: str
        """
        if self.data_dir is None:
            raise ValueError("trace data not (yet) loaded")
        if self.ftrace and hasattr(self.ftrace, event):
            return getattr(self.ftrace, event).data_frame
        raise ValueError('Event [{}] not supported. '
                         'Supported events are: {}'
                         .format(event, self.available_events))

    def _dfg_functions_stats(self, functions=None):
        """
        Get a DataFrame of specified kernel functions profile data

        For each profiled function a DataFrame is returned which reports stats
        on kernel functions execution time. The reported stats are per-CPU and
        includes: number of times the function has been executed (hits),
        average execution time (avg), overall execution time (time) and samples
        variance (s_2).
        By default returns a DataFrame of all the functions profiled.

        :param functions: the name of the function or a list of function names
                          to report
        :type functions: str or list(str)
        """
        if not hasattr(self, '_functions_stats_df'):
            return None
        df = self._functions_stats_df
        if not functions:
            return df
        return df.loc[df.index.get_level_values(1).isin(listify(functions))]

    # cgroup_attach_task with just merged fake and real events
    def _cgroup_attach_task(self):
        cgroup_events = ['cgroup_attach_task', 'cgroup_attach_task_devlib']
        df = None

        if set(cgroup_events).isdisjoint(set(self.available_events)):
            self._log.error('atleast one of {} is needed for cgroup_attach_task event generation'.format(cgroup_events))
            return None

        for cev in cgroup_events:
            if not cev in self.available_events:
                continue
            cdf = self._dfg_trace_event(cev)
            cdf = cdf[['__line', 'pid', 'controller', 'cgroup']]
            if not isinstance(df, pd.DataFrame):
                df = cdf
            else:
                df = pd.concat([cdf, df])

        # Always drop na since this DF is used as secondary
        df.dropna(inplace=True, how='any')
        return df

    @memoized
    def _dfg_cgroup_attach_task(self, controllers = ['schedtune', 'cpuset']):
        # Since fork doesn't result in attach events, generate fake attach events
        # The below mechanism doesn't work to propogate nested fork levels:
        # For ex:
        # cgroup_attach_task: pid=1166
        # fork: pid=1166 child_pid=2222  <-- fake attach generated
        # fork: pid=2222 child_pid=3333  <-- fake attach not generated
        def fork_add_cgroup(fdf, cdf, controller):
            cdf = cdf[cdf['controller'] == controller]
            ret_df = trappy.utils.merge_dfs(fdf, cdf, pivot='pid')
            return ret_df

        if not 'sched_process_fork' in self.available_events:
            self._log.error('sched_process_fork is mandatory to get proper cgroup_attach events')
            return None
        fdf = self._dfg_trace_event('sched_process_fork')

        forks_len = len(fdf)
        forkdf = fdf
        cdf = self._cgroup_attach_task()
        for idx, c in enumerate(controllers):
            fdf = fork_add_cgroup(fdf, cdf, c)
            if (idx != (len(controllers) - 1)):
                fdf = pd.concat([fdf, forkdf]).sort_values(by='__line')

        fdf = fdf[['__line', 'child_pid', 'controller', 'cgroup']]
        fdf.rename(columns = { 'child_pid': 'pid' }, inplace=True)

        # Always drop na since this DF is used as secondary
        fdf.dropna(inplace=True, how='any')

        new_forks_len = len(fdf) / len(controllers)

        fdf = pd.concat([fdf, cdf]).sort_values(by='__line')

        if new_forks_len < forks_len:
            dropped = forks_len - new_forks_len
            self._log.info("Couldn't attach all forks cgroup with-attach events ({} dropped)".format(dropped))
        return fdf

    @memoized
    def _dfg_sched_switch_cgroup(self, controllers = ['schedtune', 'cpuset']):
        def sched_switch_add_cgroup(sdf, cdf, controller, direction):
            cdf = cdf[cdf['controller'] == controller]

            ret_df = sdf.rename(columns = { direction + '_pid': 'pid' })
            ret_df = trappy.utils.merge_dfs(ret_df, cdf, pivot='pid')
            ret_df.rename(columns = { 'pid': direction + '_pid' }, inplace=True)

            ret_df.drop('controller', axis=1, inplace=True)
            ret_df.rename(columns = { 'cgroup': direction + '_' + controller }, inplace=True)
            return ret_df

        if not 'sched_switch' in self.available_events:
            self._log.error('sched_switch is mandatory to generate sched_switch_cgroup event')
            return None
        sdf = self._dfg_trace_event('sched_switch')
        cdf = self._dfg_cgroup_attach_task()

        for c in controllers:
            sdf = sched_switch_add_cgroup(sdf, cdf, c, 'next')
            sdf = sched_switch_add_cgroup(sdf, cdf, c, 'prev')

        # Augment with TGID information
        sdf = sdf.join(self._pid_tgid, on='next_pid').rename(columns = {'tgid': 'next_tgid'})
        sdf = sdf.join(self._pid_tgid, on='prev_pid').rename(columns = {'tgid': 'prev_tgid'})

        df = self._tasks_by_pid.rename(columns = { 'next_comm': 'comm' })
        sdf = sdf.join(df, on='next_tgid').rename(columns = {'comm': 'next_tgid_comm'})
        sdf = sdf.join(df, on='prev_tgid').rename(columns = {'comm': 'prev_tgid_comm'})
        return sdf

###############################################################################
# Trace Events Sanitize Methods
###############################################################################

    def _sanitize_SchedCpuCapacity(self):
        """
        Add more columns to cpu_capacity data frame if the energy model is
        available.
        """
        if not self.hasEvents('cpu_capacity') \
           or 'nrg_model' not in self.platform:
            return

        df = self._dfg_trace_event('cpu_capacity')

        # Add column with LITTLE and big CPUs max capacities
        nrg_model = self.platform['nrg_model']
        max_lcap = nrg_model['little']['cpu']['cap_max']
        max_bcap = nrg_model['big']['cpu']['cap_max']
        df['max_capacity'] = np.select(
                [df.cpu.isin(self.platform['clusters']['little'])],
                [max_lcap], max_bcap)
        # Add LITTLE and big CPUs "tipping point" threshold
        tip_lcap = 0.8 * max_lcap
        tip_bcap = 0.8 * max_bcap
        df['tip_capacity'] = np.select(
                [df.cpu.isin(self.platform['clusters']['little'])],
                [tip_lcap], tip_bcap)

    def _sanitize_SchedLoadAvgCpu(self):
        """
        If necessary, rename certain signal names from v5.0 to v5.1 format.
        """
        if not self.hasEvents('sched_load_avg_cpu'):
            return
        df = self._dfg_trace_event('sched_load_avg_cpu')
        if 'utilization' in df:
            df.rename(columns={'utilization': 'util_avg'}, inplace=True)
            df.rename(columns={'load': 'load_avg'}, inplace=True)

    def _sanitize_SchedLoadAvgTask(self):
        """
        If necessary, rename certain signal names from v5.0 to v5.1 format.
        """
        if not self.hasEvents('sched_load_avg_task'):
            return
        df = self._dfg_trace_event('sched_load_avg_task')
        if 'utilization' in df:
            df.rename(columns={'utilization': 'util_avg'}, inplace=True)
            df.rename(columns={'load': 'load_avg'}, inplace=True)
            df.rename(columns={'avg_period': 'period_contrib'}, inplace=True)
            df.rename(columns={'runnable_avg_sum': 'load_sum'}, inplace=True)
            df.rename(columns={'running_avg_sum': 'util_sum'}, inplace=True)
        df['cluster'] = np.select(
                [df.cpu.isin(self.platform['clusters']['little'])],
                ['LITTLE'], 'big')
        # Add a column which represents the max capacity of the smallest
        # clustre which can accomodate the task utilization
        little_cap = self.platform['nrg_model']['little']['cpu']['cap_max']
        big_cap = self.platform['nrg_model']['big']['cpu']['cap_max']
        df['min_cluster_cap'] = df.util_avg.map(
            lambda util_avg: big_cap if util_avg > little_cap else little_cap
        )

    def _sanitize_SchedBoostCpu(self):
        """
        Add a boosted utilization signal as the sum of utilization and margin.

        Also, if necessary, rename certain signal names from v5.0 to v5.1
        format.
        """
        if not self.hasEvents('sched_boost_cpu'):
            return
        df = self._dfg_trace_event('sched_boost_cpu')
        if 'usage' in df:
            df.rename(columns={'usage': 'util'}, inplace=True)
        df['boosted_util'] = df['util'] + df['margin']

    def _sanitize_SchedBoostTask(self):
        """
        Add a boosted utilization signal as the sum of utilization and margin.

        Also, if necessary, rename certain signal names from v5.0 to v5.1
        format.
        """
        if not self.hasEvents('sched_boost_task'):
            return
        df = self._dfg_trace_event('sched_boost_task')
        if 'utilization' in df:
            # Convert signals name from to v5.1 format
            df.rename(columns={'utilization': 'util'}, inplace=True)
        df['boosted_util'] = df['util'] + df['margin']

    def _sanitize_SchedEnergyDiff(self):
        """
        If a energy model is provided, some signals are added to the
        sched_energy_diff trace event data frame.

        Also convert between existing field name formats for sched_energy_diff
        """
        if not self.hasEvents('sched_energy_diff') \
           or 'nrg_model' not in self.platform:
            return
        nrg_model = self.platform['nrg_model']
        em_lcluster = nrg_model['little']['cluster']
        em_bcluster = nrg_model['big']['cluster']
        em_lcpu = nrg_model['little']['cpu']
        em_bcpu = nrg_model['big']['cpu']
        lcpus = len(self.platform['clusters']['little'])
        bcpus = len(self.platform['clusters']['big'])
        SCHED_LOAD_SCALE = 1024

        power_max = em_lcpu['nrg_max'] * lcpus + em_bcpu['nrg_max'] * bcpus + \
            em_lcluster['nrg_max'] + em_bcluster['nrg_max']
        self._log.debug(
            "Maximum estimated system energy: {0:d}".format(power_max))

        df = self._dfg_trace_event('sched_energy_diff')

        translations = {'nrg_d' : 'nrg_diff',
                        'utl_d' : 'usage_delta',
                        'payoff' : 'nrg_payoff'
        }
        df.rename(columns=translations, inplace=True)

        df['nrg_diff_pct'] = SCHED_LOAD_SCALE * df.nrg_diff / power_max

        # Tag columns by usage_delta
        ccol = df.usage_delta
        df['usage_delta_group'] = np.select(
            [ccol < 150, ccol < 400, ccol < 600],
            ['< 150', '< 400', '< 600'], '>= 600')

        # Tag columns by nrg_payoff
        ccol = df.nrg_payoff
        df['nrg_payoff_group'] = np.select(
            [ccol > 2e9, ccol > 0, ccol > -2e9],
            ['Optimal Accept', 'SchedTune Accept', 'SchedTune Reject'],
            'Suboptimal Reject')

    def _sanitize_SchedOverutilized(self):
        """ Add a column with overutilized status duration. """
        if not self.hasEvents('sched_overutilized'):
            return
        df = self._dfg_trace_event('sched_overutilized')
        df['start'] = df.index
        df['len'] = (df.start - df.start.shift()).fillna(0).shift(-1)
        df.drop('start', axis=1, inplace=True)

    # Sanitize cgroup information helper
    def _helper_sanitize_CgroupAttachTask(self, df, allowed_cgroups, controller_id_name):
        # Drop rows that aren't in the root-id -> name map
        df = df[df['dst_root'].isin(controller_id_name.keys())]

        def get_cgroup_name(path, valid_names):
            name = os.path.basename(path)
            name = 'root' if not name in valid_names else name
            return name

        def get_cgroup_names(rows):
            ret = []
            for r in rows.iterrows():
                 ret.append(get_cgroup_name(r[1]['dst_path'], allowed_cgroups))
            return ret

        def get_controller_names(rows):
            ret = []
            for r in rows.iterrows():
                 ret.append(controller_id_name[r[1]['dst_root']])
            return ret

        # Sanitize cgroup names
        # cgroup column isn't in mainline, add it in
        # its already added for some out of tree kernels so check first
        if not 'cgroup' in df.columns:
            if not 'dst_path' in df.columns:
                raise RuntimeError('Cant santize cgroup DF, need dst_path')
            df = df.assign(cgroup = get_cgroup_names)

        # Sanitize controller names
        if not 'controller' in df.columns:
            if not 'dst_root' in df.columns:
                raise RuntimeError('Cant santize cgroup DF, need dst_path')
            df = df.assign(controller = get_controller_names)

        return df

    def _sanitize_CgroupAttachTask(self):
        def sanitize_cgroup_event(name):
            if not name in self.available_events:
                return

            df = self._dfg_trace_event(name)

            if len(df.groupby(level=0).filter(lambda x: len(x) > 1)) > 0:
                self._log.warning('Timstamp Collisions seen in {} event!'.format(name))

            df = self._helper_sanitize_CgroupAttachTask(df, self.cgroup_info['cgroups'],
                                              self.cgroup_info['controller_ids'])
            getattr(self.ftrace, name).data_frame = df
        sanitize_cgroup_event('cgroup_attach_task')
        sanitize_cgroup_event('cgroup_attach_task_devlib')

    def _chunker(self, seq, size):
        """
        Given a data frame or a series, generate a sequence of chunks of the
        given size.

        :param seq: data to be split into chunks
        :type seq: :mod:`pandas.Series` or :mod:`pandas.DataFrame`

        :param size: size of each chunk
        :type size: int
        """
        return (seq.iloc[pos:pos + size] for pos in range(0, len(seq), size))

    def _sanitize_CpuFrequency(self):
        """
        Verify that all platform reported clusters are frequency coherent (i.e.
        frequency scaling is performed at a cluster level).
        """
        if not self.hasEvents('cpu_frequency_devlib'):
            return

        devlib_freq = self._dfg_trace_event('cpu_frequency_devlib')
        devlib_freq.rename(columns={'cpu_id':'cpu'}, inplace=True)
        devlib_freq.rename(columns={'state':'frequency'}, inplace=True)

        df = self._dfg_trace_event('cpu_frequency')
        clusters = self.platform['clusters']

        # devlib always introduces fake cpu_frequency events, in case the
        # OS has not generated cpu_frequency envets there are the only
        # frequency events to report
        if len(df) == 0:
            # Register devlib injected events as 'cpu_frequency' events
            setattr(self.ftrace.cpu_frequency, 'data_frame', devlib_freq)
            df = devlib_freq
            self.available_events.append('cpu_frequency')

        # make sure fake cpu_frequency events are never interleaved with
        # OS generated events
        else:
            if len(devlib_freq) > 0:

                # Frequencies injection is done in a per-cluster based.
                # This is based on the assumption that clusters are
                # frequency choerent.
                # For each cluster we inject devlib events only if
                # these events does not overlaps with os-generated ones.

                # Inject "initial" devlib frequencies
                os_df = df
                dl_df = devlib_freq.iloc[:self.platform['cpus_count']]
                for _,c in self.platform['clusters'].iteritems():
                    dl_freqs = dl_df[dl_df.cpu.isin(c)]
                    os_freqs = os_df[os_df.cpu.isin(c)]
                    self._log.debug("First freqs for %s:\n%s", c, dl_freqs)
                    # All devlib events "before" os-generated events
                    self._log.debug("Min os freq @: %s", os_freqs.index.min())
                    if os_freqs.empty or \
                       os_freqs.index.min() > dl_freqs.index.max():
                        self._log.debug("Insert devlib freqs for %s", c)
                        df = pd.concat([dl_freqs, df])

                # Inject "final" devlib frequencies
                os_df = df
                dl_df = devlib_freq.iloc[self.platform['cpus_count']:]
                for _,c in self.platform['clusters'].iteritems():
                    dl_freqs = dl_df[dl_df.cpu.isin(c)]
                    os_freqs = os_df[os_df.cpu.isin(c)]
                    self._log.debug("Last freqs for %s:\n%s", c, dl_freqs)
                    # All devlib events "after" os-generated events
                    self._log.debug("Max os freq @: %s", os_freqs.index.max())
                    if os_freqs.empty or \
                       os_freqs.index.max() < dl_freqs.index.min():
                        self._log.debug("Append devlib freqs for %s", c)
                        df = pd.concat([df, dl_freqs])

                df.sort_index(inplace=True)

            setattr(self.ftrace.cpu_frequency, 'data_frame', df)

        # Frequency Coherency Check
        for _, cpus in clusters.iteritems():
            cluster_df = df[df.cpu.isin(cpus)]
            for chunk in self._chunker(cluster_df, len(cpus)):
                f = chunk.iloc[0].frequency
                if any(chunk.frequency != f):
                    self._log.warning('Cluster Frequency is not coherent! '
                                      'Failure in [cpu_frequency] events at:')
                    self._log.warning(chunk)
                    self.freq_coherency = False
                    return
        self._log.info('Platform clusters verified to be Frequency coherent')

###############################################################################
# Utility Methods
###############################################################################

    def integrate_square_wave(self, sq_wave):
        """
        Compute the integral of a square wave time series.

        :param sq_wave: square wave assuming only 1.0 and 0.0 values
        :type sq_wave: :mod:`pandas.Series`
        """
        sq_wave.iloc[-1] = 0.0
        # Compact signal to obtain only 1-0-1-0 sequences
        comp_sig = sq_wave.loc[sq_wave.shift() != sq_wave]
        # First value for computing the difference must be a 1
        if comp_sig.iloc[0] == 0.0:
            return sum(comp_sig.iloc[2::2].index - comp_sig.iloc[1:-1:2].index)
        else:
            return sum(comp_sig.iloc[1::2].index - comp_sig.iloc[:-1:2].index)

    def _loadFunctionsStats(self, path='trace.stats'):
        """
        Read functions profiling file and build a data frame containing all
        relevant data.

        :param path: path to the functions profiling trace file
        :type path: str
        """
        if os.path.isdir(path):
            path = os.path.join(path, 'trace.stats')
        if path.endswith('dat') or path.endswith('html'):
            pre, ext = os.path.splitext(path)
            path = pre + '.stats'
        if not os.path.isfile(path):
            return False

        # Opening functions profiling JSON data file
        self._log.debug('Loading functions profiling data from [%s]...', path)
        with open(os.path.join(path), 'r') as fh:
            trace_stats = json.load(fh)

        # Build DataFrame of function stats
        frames = {}
        for cpu, data in trace_stats.iteritems():
            frames[int(cpu)] = pd.DataFrame.from_dict(data, orient='index')

        # Build and keep track of the DataFrame
        self._functions_stats_df = pd.concat(frames.values(),
                                             keys=frames.keys())

        return len(self._functions_stats_df) > 0

    @memoized
    def getCPUActiveSignal(self, cpu):
        """
        Build a square wave representing the active (i.e. non-idle) CPU time,
        i.e.:

          cpu_active[t] == 1 if the CPU is reported to be non-idle by cpuidle at
          time t
          cpu_active[t] == 0 otherwise

        :param cpu: CPU ID
        :type cpu: int

        :returns: A :mod:`pandas.Series` or ``None`` if the trace contains no
                  "cpu_idle" events
        """
        if not self.hasEvents('cpu_idle'):
            self._log.warning('Events [cpu_idle] not found, '
                              'cannot compute CPU active signal!')
            return None

        idle_df = self._dfg_trace_event('cpu_idle')
        cpu_df = idle_df[idle_df.cpu_id == cpu]

        cpu_active = cpu_df.state.apply(
            lambda s: 1 if s == NON_IDLE_STATE else 0
        )

        start_time = 0.0
        if not self.ftrace.normalized_time:
            start_time = self.ftrace.basetime

        if cpu_active.empty:
            cpu_active = pd.Series([0], index=[start_time])
        elif cpu_active.index[0] != start_time:
            entry_0 = pd.Series(cpu_active.iloc[0] ^ 1, index=[start_time])
            cpu_active = pd.concat([entry_0, cpu_active])

        # Fix sequences of wakeup/sleep events reported with the same index
        return handle_duplicate_index(cpu_active)


    @memoized
    def getClusterActiveSignal(self, cluster):
        """
        Build a square wave representing the active (i.e. non-idle) cluster
        time, i.e.:

          cluster_active[t] == 1 if at least one CPU is reported to be non-idle
          by CPUFreq at time t
          cluster_active[t] == 0 otherwise

        :param cluster: list of CPU IDs belonging to a cluster
        :type cluster: list(int)

        :returns: A :mod:`pandas.Series` or ``None`` if the trace contains no
                  "cpu_idle" events
        """
        if not self.hasEvents('cpu_idle'):
            self._log.warning('Events [cpu_idle] not found, '
                              'cannot compute cluster active signal!')
            return None

        active = self.getCPUActiveSignal(cluster[0]).to_frame(name=cluster[0])
        for cpu in cluster[1:]:
            active = active.join(
                self.getCPUActiveSignal(cpu).to_frame(name=cpu),
                how='outer'
            )

        active.fillna(method='ffill', inplace=True)

        # Cluster active is the OR between the actives on each CPU
        # belonging to that specific cluster
        cluster_active = reduce(
            operator.or_,
            [cpu_active.astype(int) for _, cpu_active in
             active.iteritems()]
        )

        return cluster_active


class TraceData:
    """ A DataFrame collector exposed to Trace's clients """
    pass

# vim :set tabstop=4 shiftwidth=4 expandtab