# 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 import matplotlib import numpy as np import pandas as pd import tempfile import os import warnings from test_thermal import BaseTestThermal import trappy class TestPlotter(BaseTestThermal): """No Bombing testcases for plotter""" def __init__(self, *args, **kwargs): super(TestPlotter, self).__init__(*args, **kwargs) def test_plot_no_pivot(self): """Tests LinePlot with no pivot""" trace1 = trappy.FTrace(name="first") l = trappy.LinePlot(trace1, trappy.thermal.Thermal, column="temp") l.view(test=True) def test_plot_multi_trace(self): """Tests LinePlot with no Pivot multi traces""" trace1 = trappy.FTrace(name="first") trace2 = trappy.FTrace(name="second") l = trappy.LinePlot( [trace1, trace2], trappy.thermal.Thermal, column="temp") l.view(test=True) def test_plot_multi(self): """Tests LinePlot with no Pivot multi attrs""" trace1 = trappy.FTrace(name="first") trace2 = trappy.FTrace(name="second") l = trappy.LinePlot([trace1, trace2], [trappy.thermal.Thermal, trappy.thermal.ThermalGovernor], column=["temp", "power_range"]) l.view(test=True) def test_plot_filter(self): """Tests LinePlot with no Pivot with filters""" trace1 = trappy.FTrace(name="first") trace2 = trappy.FTrace(name="second") l = trappy.LinePlot([trace1, trace2], [trappy.cpu_power.CpuOutPower], column=["power"], filters={"cdev_state": [0]}) l.view(test=True) def test_plot_pivot(self): """Tests LinePlot with Pivot""" trace1 = trappy.FTrace(name="first") l = trappy.LinePlot( trace1, trappy.thermal.Thermal, column="temp", pivot="thermal_zone") l.view(test=True) def test_plot_multi_trace_pivot(self): """Tests LinePlot with Pivot multi traces""" trace1 = trappy.FTrace(name="first") trace2 = trappy.FTrace(name="second") l = trappy.LinePlot( [trace1, trace2], trappy.cpu_power.CpuOutPower, column="power", pivot="cpus") l.view(test=True) def test_plot_multi_pivot(self): """Tests LinePlot with Pivot with multi attrs""" trace1 = trappy.FTrace(name="first") trace2 = trappy.FTrace(name="second") l = trappy.LinePlot([trace1, trace2], [trappy.cpu_power.CpuInPower, trappy.cpu_power.CpuOutPower], column=["dynamic_power", "power"], pivot="cpus") l.view(test=True) def test_plot_multi_pivot_filter(self): """Tests LinePlot with Pivot and filters""" trace1 = trappy.FTrace(name="first") trace2 = trappy.FTrace(name="second") l = trappy.LinePlot( trace1, trappy.cpu_power.CpuInPower, column=[ "dynamic_power", "load1"], filters={ "cdev_state": [ 1, 0]}, pivot="cpus") l.view(test=True) def test_plot_savefig(self): """Tests plotter: savefig""" trace1 = trappy.FTrace(name="first") trace2 = trappy.FTrace(name="second") l = trappy.LinePlot( trace1, trappy.cpu_power.CpuInPower, column=[ "dynamic_power", "load1"], filters={ "cdev_state": [ 1, 0]}, pivot="cpus") png_file = tempfile.mktemp(dir="/tmp", suffix=".png") l.savefig(png_file) self.assertTrue(os.path.isfile(png_file)) os.remove(png_file) def test_signals(self): """Test signals input for LinePlot""" trace1 = trappy.FTrace(name="first") trace2 = trappy.FTrace(name="second") l = trappy.LinePlot([trace1, trace2], signals=["cpu_in_power:dynamic_power", "cpu_out_power:power"], pivot="cpus") self.assertTrue(isinstance(l.templates[0], type(trappy.cpu_power.CpuInPower))) self.assertEquals(l._attr["column"][0], "dynamic_power") self.assertTrue(l.templates[1], type(trappy.cpu_power.CpuOutPower)) self.assertEquals(l._attr["column"][1], "power") self.assertTrue("colors" not in l._attr) # Check that plotting doesn't barf l.view(test=True) def test_signals_exceptions(self): """Test incorrect input combinations: signals""" trace1 = trappy.FTrace(name="first") trace2 = trappy.FTrace(name="second") with self.assertRaises(ValueError): l = trappy.LinePlot([trace1, trace2], column=[ "dynamic_power", "load1"], signals=["cpu_in_power:dynamic_power", "cpu_out_power:power"], pivot="cpus") with self.assertRaises(ValueError): l = trappy.LinePlot([trace1, trace2], trappy.cpu_power.CpuInPower, signals=["cpu_in_power:dynamic_power", "cpu_out_power:power"], pivot="cpus") with self.assertRaises(ValueError): l = trappy.LinePlot([trace1, trace2], trappy.cpu_power.CpuInPower, column=[ "dynamic_power", "load1"], signals=["cpu_in_power:dynamic_power", "cpu_out_power:power"], pivot="cpus") def test_signals_invalid(self): """Test that invalid signal defs result in a helpful errror""" trace = trappy.FTrace() with self.assertRaises(ValueError) as assertion: l = trappy.LinePlot(trace, signals=["INVALID_SIGNAL_DEF"]) msg = str(assertion.exception) self.assertIn("Invalid signal definition", msg) self.assertIn("INVALID_SIGNAL_DEF", msg) def test_signals_colors(self): """Test signals with colors in LinePlot""" trace1 = trappy.FTrace(name="first") trace2 = trappy.FTrace(name="second") l = trappy.LinePlot([trace1, trace2], signals=["thermal:temp:1,2,3", "cpu_in_power:load2:200,100,0"], pivot="cpus") self.assertTrue(isinstance(l.templates[0], type(trappy.thermal.Thermal))) self.assertEquals(l._attr["column"][0], "temp") self.assertEquals(l._attr["colors"][0], [1, 2, 3]) self.assertTrue(l.templates[1], type(trappy.cpu_power.CpuInPower)) self.assertEquals(l._attr["column"][1], "load2") self.assertEquals(l._attr["colors"][1], [200, 100, 0]) # Check that plotting doesn't barf l.view(test=True) # Test hex color l = trappy.LinePlot([trace1, trace2], signals=["thermal:prev_temp:0xff,0x3a,0x3"], pivot="cpus") self.assertEquals(l._attr["colors"][0], [0xff, 0x3a, 0x3]) def test_lineplot_dataframe(self): """LinePlot plots DataFrames without exploding""" data = np.random.randn(4, 2) dfr = pd.DataFrame(data, columns=["tick", "tock"]).cumsum() trappy.LinePlot(dfr, column=["tick"]).view(test=True) def test_get_trace_event_data_corrupted_trace(self): """get_trace_event_data() works with a corrupted trace""" from trappy.plotter.Utils import get_trace_event_data trace = trappy.FTrace() # We create this trace: # # 1 15414 -> 15411 # 2 15411 -> 15414 # 3 15414 -> 15411 (corrupted, should be dropped) # 4 15413 -> 15411 # 5 15411 -> 15413 # # Which should plot like: # # CPU # +-------+-------+ # 0 | 15411 | 15414 | # +-------+-------+ +-------+ # 1 | 15411 | # +-------+ # +-------+-------+-------+-------+ # 0.1 0.2 0.3 0.4 0.5 broken_trace = pd.DataFrame({ '__comm': ["task2", "task1", "task2", "task3", "task1"], '__cpu': [0, 0, 0, 1, 1], '__pid': [15414, 15411, 15414, 15413, 15411], 'next_comm': ["task1", "task2", "task1", "task1", "task3"], 'next_pid': [15411, 15414, 15411, 15411, 15413], 'prev_comm': ["task2", "task1", "task2", "task3", "task1"], 'prev_pid': [15414, 15411, 15414, 15413, 15411], 'prev_state': ["S", "R", "S", "S", "S"]}, index=pd.Series(range(1, 6), name="Time")) trace.sched_switch.data_frame = broken_trace with warnings.catch_warnings(record=True) as warn: data, procs, window = get_trace_event_data(trace) self.assertEquals(len(warn), 1) warn_str = str(warn[-1]) self.assertTrue("15411" in warn_str) self.assertTrue("4" in warn_str) zipped_comms = zip(broken_trace["next_comm"], broken_trace["next_pid"]) expected_procs = set("-".join([comm, str(pid)]) for comm, pid in zipped_comms) self.assertTrue([1, 2, 0] in data["task1-15411"]) self.assertTrue([2, 3, 0] in data["task2-15414"]) self.assertTrue([4, 5, 1] in data["task1-15411"]) self.assertEquals(procs, expected_procs) self.assertEquals(window, [1, 5]) class TestILinePlotter(unittest.TestCase): def test_simple_dfr(self): """ILinePlot doesn't barf when plotting DataFrames""" dfr1 = pd.DataFrame([1, 2, 3, 4], columns=["a"]) dfr2 = pd.DataFrame([2, 3, 4, 5], columns=["a"]) trappy.ILinePlot([dfr1, dfr2], column=["a", "a"]).view(test=True) with self.assertRaises(ValueError): trappy.ILinePlot([dfr1, dfr2]).view(test=True) def test_df_to_dygraph(self): """Test the ILinePlot util function: df_to_dygraph""" dfr1 = pd.DataFrame([[1, 2], [3, 4], [5, 6]], index=[0., 1., 2.], columns=["a", "b"]) dfr2 = pd.DataFrame([1, 2, 3, 4], index=[0., 1., 2., 3.], columns=["a"]) expected_result_1 = { 'labels': ['index', 'a', 'b'], 'data': [[0.0, 1, 2], [1.0, 3, 4], [2.0, 5, 6]] } expected_result_2 = { 'labels': ['index', 'a'], 'data': [[0.0, 1], [1.0, 2], [2.0, 3], [3.0, 4]] } result_1 = trappy.plotter.ILinePlotGen.df_to_dygraph(dfr1) result_2 = trappy.plotter.ILinePlotGen.df_to_dygraph(dfr2) self.assertDictEqual(result_1, expected_result_1) self.assertDictEqual(result_2, expected_result_2) def test_duplicate_merging(self): dfr1 = pd.DataFrame([1, 2, 3, 4], index=[0., 0., 1., 2.], columns=["a"]) dfr2 = pd.DataFrame([2, 3, 4, 5], index=[1., 1., 1., 2.], columns=["a"]) trappy.ILinePlot([dfr1, dfr2], column=["a", "a"]).view(test=True) def test_independent_series_merging(self): """ILinePlot fixes indexes of independent series""" index1 = [0., 1., 2., 3.] s1 = pd.Series([1, 2, 3, 4], index=index1) index2 = [0.5, 1.5, 2.5, 3.5] s2 = pd.Series([2, 3, 4, 5], index=index2) dfr = pd.DataFrame([0, 1, 2, 3], columns=["a"]) iplot = trappy.ILinePlot(dfr, column=["a"]) s = {"s1": s1, "s2": s2} merged = iplot._fix_indexes(s) expected_index = index1 + index2 expected_index.sort() self.assertEquals(expected_index, sorted(merged["s1"].keys())) def test_dygraph_colors(self): """Check that to_dygraph_colors() constructs a valid dygraph colors argument""" from trappy.plotter.ColorMap import to_dygraph_colors color_map = [[86, 58, 206]] expected = '["rgb(86, 58, 206)"]' self.assertEquals(to_dygraph_colors(color_map), expected) color_map = [[0, 0, 0], [123, 23, 45]] expected = '["rgb(0, 0, 0)", "rgb(123, 23, 45)"]' self.assertEquals(to_dygraph_colors(color_map), expected) class TestBarPlot(unittest.TestCase): def setUp(self): self.dfr = pd.DataFrame({"foo": [1, 2, 3], "bar": [2, 3, 1], "baz": [3, 2, 1]}) def test_barplot_dfr(self): """BarPlot plots dataframes without exploding""" trappy.BarPlot(self.dfr, column=["foo", "bar"]).view(test=True) def test_barplot_trace(self): """BarPlot plots traces without exploding""" trace = trappy.BareTrace() trace.add_parsed_event("event", self.dfr) trappy.BarPlot(trace, signals=["event:foo", "event:bar"]).view(test=True)