"""
A collection of utility function for data manipulation and plotting, built
on top of Pandas data frames and the `chart` and `ideplot` packages from
`omnetpp.scave`. Functions in this module have been written largely to the
needs of the chart templates that ship with the IDE.
There are some functions which are (almost) mandatory elements in a chart script.
These are the following.
If you want style settings in the chart dialog to take effect:
- `preconfigure_plot()`
- `postconfigure_plot()`
If you want image/data export to work:
- `export_image_if_needed()`
- `export_data_if_needed()`
"""
import logging
import random, sys, os, string, re, math
import numpy as np
import scipy.stats as st
import pandas as pd
import itertools as it
import matplotlib as mpl
import matplotlib.pyplot as plt
from omnetpp.scave import chart, ideplot, vectorops, delta_measurement
from matplotlib.colors import ListedColormap, BoundaryNorm
[docs]
logger = logging.getLogger(__name__)
[docs]
verbose_export = os.getenv("WITHIN_OMNETPP_IDE") == "yes" # use False as default, but we want verbose export in IDE
[docs]
def set_verbose_export(v):
"""
Sets the `verbose_export` flag, which controls whether the
export_image_if_needed() and export_data_if_needed() functions
will print an "Exported <filename>" message after the export.
The default setting is `False`.
"""
global verbose_export
verbose_export = v
def _import_scave_bindings():
module_suffixes = ["", "_dbg", "_sanitize"]
import importlib
sb = None
errors = []
for suffix in module_suffixes:
modulename = "omnetpp.scave.scave_bindings" + suffix
try:
sb = importlib.import_module(modulename)
logger.debug(f"Loaded {modulename}")
break
except ImportError as e:
logger.debug(f"Failed to load {modulename}: {e}")
errors.append(e)
if sb is None:
import importlib.machinery
scave_dir = os.path.dirname(os.path.abspath(__file__))
ext_suffix = importlib.machinery.EXTENSION_SUFFIXES[0]
lines = []
for s, err in zip(module_suffixes, errors):
fpath = os.path.join(scave_dir, "scave_bindings" + s + ext_suffix)
if os.path.exists(fpath):
lines.append(f" {fpath} -- found but failed to load: {err}")
else:
lines.append(f" {fpath} -- not found")
raise ImportError(
"Could not import omnetpp.scave.scave_bindings -- "
"OMNeT++ may not be fully built, or a required shared library is missing.\n"
"Files checked:\n" + "\n".join(lines)
)
return sb
[docs]
sb = _import_scave_bindings()
from ._version import __version__
def _parse_version(v: str):
import re
m = re.search(R"(\d+)\.(\d+)\.(\d+).*", v)
return int(m.group(1)), int(m.group(2)), int(m.group(3))
def _version_less_than(actual, required):
major, minor, patch = _parse_version(actual)
rmajor, rminor, rpatch = _parse_version(required)
if major != rmajor:
return major < rmajor
if minor != rminor:
return minor < rminor
return patch < rpatch
def _check_version(module, required):
if _version_less_than(module.__version__, required):
logger.warning(f"'{module.__name__}' is too old, some analysis tool functionality may be broken. "
f"Required: {required}, present: {module.__version__}. "
f"Try running `python3 -m pip install --user --upgrade {module.__name__}` to upgrade.")
_check_version(np, "1.18.0") # Dec 22, 2019
_check_version(pd, "1.0.0") # January 29, 2020
# This bug was triggered by `_plot_enum()`, fixed in 3.5.3:
# https://github.com/matplotlib/matplotlib/issues/21915
_check_version(mpl, "3.5.3") # Aug 11, 2022
# color and marker cycles, with defaults in case _initialize_cycles() is not called
_marker_cycle = it.cycle(list("osv^<>pDd"))
_color_cycle = it.cycle(["C" + str(i) for i in range(10)])
# Because str.removeprefix was only added in Python 3.9
def _removeprefix(str, prefix):
return str[len(prefix):] if str.startswith(prefix) else str
# There are many potential ways to add digit grouping to numbers.
# None of them were good enough:
# - They all can only use "," or "_" as group separators
# - And none of them added grouping to the fractional part
# - Or they printed numbers with silly precision,
# either fixed, or too low, or too high.
# So, instead, take the built-in tick labels, and add digit grouping
# to them manually, after the fact, using good ol' regices.
class _DigitGroupingFormatter(mpl.ticker.ScalarFormatter):
def __call__(self, x, pos=None):
SEP = '\u2009' # Unicode "Thin Space"
result = super().__call__(x, pos)
originalresult = result
if "." in result:
# For numbers with decimal separators in them:
# First replace any digit followed by any number of three-digit-groups,
# and a period (so, the whole part); with itself followed by a space.
result = re.sub(r'(\d)(?=(\d{3})+\.)', r'\1' + SEP, result)
# Then replace any remaining three-digit-groups (in the fractional part)
# with itself followed by a space.
result = re.sub(r'(\d{3})(?=\d)', r'\1' + SEP, result)
else:
# For integers, do the same thing as with the whole part of fractions,
# but don't require the period at the end.
result = re.sub(r'(\d)(?=(\d{3})+(?!\d))', r'\1' + SEP, result)
assert(result.replace(SEP, "") == originalresult)
# Add back the exponent to every value manually, as we don't want to see it
# extracted into a common order of magnitude, which is disabled by `get_offset()`.
# (except if it has no digits in it other than 0)
if self.orderOfMagnitude != 0 and re.search(r'[1-9]', result):
result += "e" + str(self.orderOfMagnitude)
return result
# The axis-common order of magnitude (if scientific notation is used) is also returned
# in this, but we append it to each individual tick label instead.
def get_offset(self):
return ""
def _check_same_unit(df):
"""
Checks whether the results in the passed DataFrame all have the same unit.
If so, returns it. Otherwise, raises an exception.
If the DataFrame is empty, or has no "unit" column, or if all units
are either `None`, `NaN`, or empty string, then `None` is returned.
"""
if df is None or len(df) == 0 or "unit" not in df.columns:
return None
def is_nonunit(u):
nonunits = [None, math.nan, np.nan, ""]
return u in nonunits or (isinstance(u, float) and math.isnan(u))
unit_col = df["unit"]
# The units column, but with all values in it that are considered non-units replaced with None.
unified_nonunits = unit_col.mask(unit_col.map(is_nonunit), None)
units = list(unified_nonunits.unique())
if not units:
return None
if len(units) > 1:
units = ["<none>" if u is None else u for u in units]
raise chart.ChartScriptError("The data frame contains multiple units: " + ", ".join(units))
return units[0]
def _get_best_unit(values, current_unit):
# Handle empty arrays by filtering them out (NumPy min/max doesn't like them)
non_empty_values = values[values.apply(lambda x: not isinstance(x, np.ndarray) or x.size > 0)]
if non_empty_values.empty:
return current_unit
min_value = non_empty_values.apply(np.min).min()
max_value = non_empty_values.apply(np.max).max()
mid_value = (min_value + max_value) / 2
return uc.getBestUnit(mid_value, current_unit)
def _convert_to_unit(values, original_unit, target_unit):
if original_unit == target_unit:
return values
def conv(v):
if isinstance(v, np.ndarray):
clone = v.copy()
uc.convertUnitArray(clone, original_unit, target_unit)
return clone
else:
return uc.convertUnit(v, original_unit, target_unit)
return values.apply(conv)
[docs]
def convert_to_base_unit(df, columns_to_convert=["value", "min", "max", "mean", "stddev", "vecvalue", "binedges"]):
"""
Converts results with units in the passed DataFrame to their base units in-place.
The DataFrame needs to have a "unit" column - which is updated to the base unit.
By default, the following columns are converted:
"value", "min", "max", "mean", "stddev", "vecvalue", "binedges"
Every converted column must contain either all numbers or all `np.ndarray` instances.
This works for example on the DataFrames returned by `get_scalars`, `get_vectors`, `get_statistics`,
and `get_histograms` in `omnetpp.scave`, but NOT on those returned by `get_results`.
"""
if "unit" not in df.columns:
raise chart.ChartScriptError("The DataFrame does not contain a 'unit' column")
if df.empty:
return
# Precompute eligible columns and separate them by dtype
numeric_columns = [col for col in df.columns
if col in columns_to_convert
and pd.api.types.is_numeric_dtype(df[col])]
ndarray_columns = [col for col in df.columns
if col in columns_to_convert
and isinstance(df[col].iloc[0], np.ndarray)]
# Iterate through rows
for index, row in df.iterrows():
original_unit = row["unit"]
if original_unit is None or original_unit == "" or not isinstance(original_unit, str):
continue
# Convert unit for the current row
base_unit = uc.getBaseUnit(original_unit)
if base_unit is None or base_unit == "" or base_unit == original_unit:
continue
df.at[index, "unit"] = base_unit
# Convert numeric quantity columns to base unit
for col in numeric_columns:
value = row[col]
converted_value = uc.convertUnit(value, original_unit, base_unit)
df.at[index, col] = converted_value
# Convert ndarray quantity columns to base unit
for col in ndarray_columns:
values = row[col].copy() # NOTE: I don't know why this copy is needed...
uc.convertUnitArray(values, original_unit, base_unit)
df.at[index, col] = values
# Note: must be at the top, because it appears in other functions' arg list as default
[docs]
def make_legend_label(legend_cols, row, props={}):
"""
Produces a reasonably good label text (to be used in a chart legend) for a result row from
a DataFrame. The legend label is produced as follows.
First, a base version of the legend label is produced:
1. If the DataFrame contains a `legend` column, its content is used.
2. Otherwise, if there is a `legend_format` property, it is used as a format string
for producing the legend label. The format string may contain references to other
columns of the DataFrame in the "$name" or "${name}" form.
3. Otherwise, the legend label is concatenated from the columns listed in `legend_cols`,
a list whose contents are usually produced using the `extract_label_columns()` function.
Second, if there is a `legend_replacements` property, it defines a series of
replacements to be done on the legend labels. `legend_replacements` is
expected to be a multi-line string, where each line contains a replacement.
A replacement line can be a plain substring replacement in the
`findstring --> replacement` syntax, or a regex replacement in the sed-style
`/findregex/replacement/` syntax. With the latter, "findregex" should be
a valid regular expression, and "replacement" a string that may contain
match group references ("\1", "\2", etc.). If "/" is unsuitable as
separator, other characters may also be used; common choices are "|" and
"!". Similar to the `legend_format` property, "findstring", "findregex" and
"replacement" may contain column references in the "$name" or "${name}"
form. Use "$$" to mean a single "$" sign. Also note that "findregex" may
still end in "$" to match the end of the string; it won't collide with
column references.
Parameters:
- `row` (named tuple): The row from the dataframe.
- `props` (dict): The properties that control how the legend is produced.
- `legend_cols` (list of strings): The names of columns chosen for the legend.
Properties that affect the generated legend label:
- `legend_automatic` (string): If `true`, do not use the legend format string even if present.
- `legend_format` (string): A format string to produce the label from columns.
- `legend_replacements` (string): A multi-line string of regex find/replace operations to modify the label.
Possible errors:
- References to nonexistent columns in `legend_format` or `legend_replacements` (`KeyError`)
- Malformed regex in the "findstring" parts of `legend_replacements` (`re.error`)
- Invalid group reference in the "replacement" parts of `legend_replacements` (`re.error`)
"""
if hasattr(row, 'legend'):
return row.legend
def get_prop(k):
return props[k] if k in props else None
def substitute_columns(format, row, where):
try:
return string.Template(format).substitute(row.to_dict() if isinstance(row, pd.Series) else row._asdict())
except KeyError as ex:
raise chart.ChartScriptError("Unknown column reference ${} {}".format(ex.args[0], where)) from ex
except ValueError as ex:
raise chart.ChartScriptError("Error {}: {}".format(where, ex.args[0])) from ex
legend_isautomatic = get_prop('legend_automatic')
legend_format = get_prop('legend_format')
if legend_format and legend_isautomatic != "true":
legend = substitute_columns(legend_format, row, "in legend format string")
else:
def fmt(col):
if col in ["name", "title"]:
prefix = ""
elif col in ["module", "moduledisplaypath"]:
prefix = " in "
else:
prefix = ", " + col + "="
return prefix + str(getattr(row, col))
legend = "".join([fmt(col) for col in legend_cols])
legend = _removeprefix(_removeprefix(legend, ", "), " in ")
legend_replacements = get_prop('legend_replacements')
if legend_replacements:
lines = [li.strip() for li in legend_replacements.split('\n') if li.strip()]
for line in lines:
if line.startswith("#"):
continue
m = re.search(r"^(\W)(.*)\1(.*)\1$", line)
if m:
findstr = m.group(2)
findstr = re.sub(r'\$$', '$$', findstr) # if ends with "$" (regex end-of-line), make it "$$" so that string.Template won't mistake it for invalid variable reference
findstr = substitute_columns(findstr, row, f"in legend label replacement `{line}`")
replacement = substitute_columns(m.group(3), row, f"in legend label replacement `{line}`")
try:
legend = re.sub(findstr, replacement, legend)
except re.error as ex:
raise chart.ChartScriptError("Regex error in legend replacement '{}': {}".format(line, str(ex))) from ex
elif "-->" in line:
(findstr, replacement) = line.split("-->", 1)
findstr = substitute_columns(findstr.strip(), row, f"in legend label replacement `{line}`")
replacement = substitute_columns(replacement.strip(), row, f"in legend label replacement `{line}`")
legend = legend.replace(findstr, replacement)
else:
raise chart.ChartScriptError("Invalid line in 'legend_replacements', 'foo --> bar' or '/foo-regex/bar-regex/' syntax expected: '" + line + "'")
return legend
[docs]
def add_legend_labels(df, props, legend_cols=None):
"""
Adds a `legend` column to the dataframe. In the dataframe, each row is
expected to represent an item to be plotted. The legend label will be
computed for each item individually by the `make_legend_label()` function.
Parameters:
- `df`: The dataframe.
- `props` (dict): The properties.
Notable properties that affect the legend generation: See the documentation
of `make_legend_label()`.
"""
if legend_cols is None:
_, legend_cols = extract_label_columns(df, props)
def get_legend_label(row):
return make_legend_label(legend_cols, row, props)
df['legend'] = df.apply(get_legend_label, axis=1)
return df
[docs]
def sort_rows_by_legend(df, props=()):
"""
Sorts the rows of the dataframe, where each row represents an item to be
plotted. The dataframe is expected to have a `legend` column, which will
serve as the basis for ordering.
Ordering is based on two lists of regexes, one for primary ordering and
another one for secondary ordering. Each item's rank will be determined by
the index of the first regex the item's legend matches. After sorting, items
matching the first regex will appear at the top, those matching the second
regex will be placed below, and so forth. Case-sensitive substring match is
used.
Parameters:
- `df`: The dataframe.
- `props` (dict): The properties.
Notable properties that affect the ordering:
- `ordering_regex_list`: Regex list for primary ordering, as multi-line string (one regex per line).
- `secondary_ordering_regex_list`: Regex list for secondary ordering, as multi-line string (one regex per line).
- `sorting`: Boolean to determine if sorting should be applied
"""
def sort_by_regex_list(df, regex_list):
regex_list = [re.compile(r) for r in regex_list if r]
if regex_list:
def find_first_match(row):
for id, regex in enumerate(regex_list):
if regex.search(row['legend']): # this is substring match
return id
return math.inf
df['order'] = df.apply(find_first_match, axis=1)
df.sort_values(by='order', kind='stable', inplace=True)
if props.get("sorting") == "true":
df.sort_values(by='legend', kind='stable', inplace=True) #TODO use natural sorting: https://saturncloud.io/blog/how-to-naturally-sort-pandas-dataframe/
if props.get("secondary_ordering_regex_list"):
sort_by_regex_list(df, props["secondary_ordering_regex_list"].split("\n"))
if props.get("ordering_regex_list"):
sort_by_regex_list(df, props["ordering_regex_list"].split("\n"))
return df
[docs]
def plot_bars(df, errors_df=None, meta_df=None, props={}, sort=True):
"""
Creates a bar plot from the dataframe, with styling and additional input
coming from the properties. Each row in the dataframe defines a series.
Group names (displayed on the x axis) are taken from the column index.
Error bars can be drawn by providing an extra dataframe of identical
dimensions as the main one. Error bars will protrude by the values in the
errors dataframe both up and down (i.e. range is 2x error).
To make the legend labels customizable, an extra dataframe can be provided,
which contains any columns of metadata for each series.
Colors are assigned automatically. The `cycle_seed` property allows you
to select other combinations if the default one is not suitable.
Parameters:
- `df`: The dataframe.
- `errors_df`: Dataframe with the errors (in y axis units).
- `meta_df`: Dataframe with the metadata about each series.
- `props` (dict): The properties.
- `sort` (bool): Whether to sort the values by the column and row indices
(which are the labels of the bar series and groups).
Notable properties that affect the plot:
- `baseline`: The y value at which the x axis is drawn.
- `bar_placement`: Selects the arrangement of bars: aligned, overlap, stacked, etc.
- `xlabel_rotation`: Amount of counter-clockwise rotation of x axis labels a.k.a. group names, in degrees.
- `title`: Plot title (autocomputed if missing).
- `cycle_seed`: Alters the sequence in which colors are assigned to series.
- `unit`: If present, it is required to be the same for all series, and it will be used in the automatic y axis label.
"""
y_unit = _check_same_unit(meta_df)
p = ideplot if chart.is_native_chart() else plt
def get_prop(k):
return props[k] if k in props else None
group_fill_ratio = 0.8
aligned_bar_fill_ratio = 0.9
overlap_visible_fraction = 1 / 3
# used only by the mpl charts
xs = np.arange(len(df.columns), dtype=np.float64) # the x locations for the groups
width = group_fill_ratio / len(df.index) # the width of the bars
bottoms = np.zeros_like(xs)
stacks = np.zeros_like(xs)
group_increment = 0.0
baseline = get_prop("baseline")
if baseline:
if ideplot.is_native_plot(): # is this how this should be done?
ideplot.set_property("Bars.Baseline", baseline)
else:
bottoms += float(baseline)
extra_args = dict()
placement = get_prop("bar_placement") or "Aligned"
if placement:
if ideplot.is_native_plot(): # is this how this should be done?
ideplot.set_property("Bar.Placement", placement)
else:
if placement == "InFront":
width = group_fill_ratio
elif placement == "Stacked":
width = group_fill_ratio
bottoms *= 0 # doesn't make sense
elif placement == "Aligned":
width = group_fill_ratio / len(df.index)
group_increment = width
xs -= width * (len(df.index)-1)/2
width *= aligned_bar_fill_ratio
elif placement == "Overlap":
width = group_fill_ratio / (1 + len(df.index) * overlap_visible_fraction)
group_increment = width * overlap_visible_fraction
extra_parts = (1.0 / overlap_visible_fraction - 1)
xs += width / extra_parts - (len(df.index) + extra_parts) * width * overlap_visible_fraction / 2
if sort:
df.sort_index(axis="columns", inplace=True)
df.sort_index(axis="index", inplace=True)
# Y unit
target_y_unit = get_prop("yaxis_unit")
if target_y_unit is None:
target_y_unit = y_unit
elif target_y_unit == "" and y_unit:
combined_values = pd.concat([df[col] for col in df.columns])
target_y_unit = _get_best_unit(combined_values, y_unit)
if y_unit != target_y_unit:
if y_unit:
for col in df.columns:
df[col] = _convert_to_unit(df[col], y_unit, target_y_unit)
y_unit = target_y_unit
if errors_df is not None:
assert df.shape == errors_df.shape
if not df.columns.equals(errors_df.columns):
# check whether the values in the column headers are the same
# in the data and error DataFrames (in perhaps different order)
assert set(df.columns) == set(errors_df.columns)
# check whether the values in the column headers are unique
# in both the data and error DataFrames
assert len(df.columns) == len(df.columns.unique())
assert len(errors_df.columns) == len(errors_df.columns.unique())
# put the errors_df columns into the same order as in the data df
def column_key(col):
return [list(df.columns).index(c) for c in col]
errors_df.sort_index(axis="columns", inplace=True, key=column_key)
assert df.columns.equals(errors_df.columns)
# Now the same again, but with rows instead of columns.
if not df.index.equals(errors_df.index):
assert set(df.index) == set(errors_df.index)
assert len(df.index) == len(df.index.unique())
assert len(errors_df.index) == len(errors_df.index.unique())
def index_key(index):
return [list(df.index).index(i) for i in index]
errors_df.sort_index(axis="index", inplace=True, key=index_key)
assert df.index.equals(errors_df.index)
title_cols, legend_cols = extract_label_columns(meta_df.reset_index(), props)
for i, ((index, row), meta_row) in enumerate(zip(df.iterrows(), meta_df.reset_index().itertuples(index=False))):
style = _make_bar_args(props, df)
if not chart.is_native_chart():
extra_args['zorder'] = 1 - (i / len(df.index) / 10)
extra_args['bottom'] = bottoms + stacks
label = make_legend_label(legend_cols, meta_row, props)
ys = row.values
p.bar(xs, ys-bottoms, width, label=label, **extra_args, **style)
if not chart.is_native_chart() and errors_df is not None and not errors_df.iloc[i].isna().all():
plt.errorbar(xs, ys + stacks, yerr=errors_df.iloc[i], capsize=float(get_prop("cap_size") or 4), **style, linestyle="none", ecolor=mpl.rcParams["axes.edgecolor"])
xs += group_increment
if placement == "Stacked":
stacks += row.values
rotation = get_prop("xlabel_rotation")
if rotation:
rotation = float(rotation)
else:
rotation = 0
p.xticks(list(range(len(df.columns))), list([_to_label(i) for i in df.columns.values]), rotation=rotation)
# Add some text for labels, title and custom x-axis tick labels, etc.
groups = df.columns.names
if "yaxis_unit" in props:
_set_xlabel(p, props, None, _to_label(groups))
_set_xlimits(p, props, None)
else:
# backward compatibility for old charts that don't have the yaxis_unit property yet
p.xlabel(_to_label(groups))
title = get_prop("title")
ylabel = ""
if meta_df is not None:
meta_df = meta_df.reset_index()
if get_prop("legend_prefer_result_titles") == "true" and "title" in meta_df:
series = meta_df["title"]
else:
series = meta_df["name"]
title_names = series.unique()
ylabel = title_names[0]
if len(title_names) > 1:
ylabel += ", etc."
if "yaxis_unit" not in props:
# backward compatibility for old charts that don't have the yaxis_unit property yet
if y_unit is not None and len(y_unit) > 0:
ylabel += f" [{y_unit}]"
p.ylabel(ylabel)
if title is None:
title = make_chart_title(meta_df, title_cols)
if "yaxis_unit" in props:
_set_ylabel(p, props, y_unit, ylabel)
_set_ylimits(p, props, y_unit)
if title is not None:
set_plot_title(title)
[docs]
def plot_vectors(df, props, legend_func=make_legend_label, sort=True):
"""
Creates a line plot from the dataframe, with styling and additional input
coming from the properties. Each row in the dataframe defines a series.
Colors and markers are assigned automatically. The `cycle_seed` property
allows you to select other combinations if the default one is not suitable.
A function to produce the legend labels can be passed in. By default,
`make_legend_label()` is used, which offers many ways to influence the
legend via dataframe columns and chart properties. In the absence of
more specified settings, the legend is normally computed from columns which best
differentiate among the vectors.
Parameters:
- `df`: The dataframe.
- `props` (dict): The properties.
- `legend_func` (function): The function to produce custom legend labels.
See `utils.make_legend_label()` for prototype and semantics.
- `sort` (bool): Whether to sort the vectors by the columns used for the legend
(before applying `legend_func`, for backwards bug-compatibility).
Columns of the dataframe:
- `vectime`, `vecvalue` (Numpy `ndarray`'s of matching sizes): the x and y coordinates for the plot
- `interpolationmode` (str, optional): this column normally comes from a result attribute, and determines how the points will be connected
- `legend` (optional): legend label for the series; if missing, legend labels are derived from other columns
- `name`, `title`, `module`, etc. (optional): provide input for the legend
Notable properties that affect the plot:
- `title`: plot title (autocomputed if missing)
- `drawstyle`: Matplotlib draw style; if present, it overrides the draw style derived from `interpolationmode`.
- `linestyle`, `linecolor`, `linewidth`, `marker`, `markersize`: styling
- `cycle_seed`: Alters the sequence in which colors and markers are assigned to series.
- `unit`: If present, it is required to be the same for all series, and it will be used in the automatic y axis label.
"""
x_unit = "s" # vectime is always simtime
y_unit = _check_same_unit(df)
p = ideplot if chart.is_native_chart() else plt
def get_prop(k):
return props[k] if k in props else None
title_cols, legend_cols = extract_label_columns(df, props)
if sort:
df.sort_values(by=legend_cols, inplace=True)
# X unit
target_x_unit = get_prop("xaxis_unit")
if target_x_unit is None:
target_x_unit = x_unit
elif target_x_unit == "" and x_unit:
target_x_unit = _get_best_unit(df.vectime, x_unit)
if x_unit != target_x_unit and x_unit:
df.vectime = _convert_to_unit(df.vectime, x_unit, target_x_unit)
x_unit = target_x_unit
# Y unit
target_y_unit = get_prop("yaxis_unit")
if target_y_unit is None:
target_y_unit = y_unit
elif target_y_unit == "" and y_unit:
target_y_unit = _get_best_unit(df.vecvalue, y_unit)
if y_unit != target_y_unit and y_unit:
df.vecvalue = _convert_to_unit(df.vecvalue, y_unit, target_y_unit)
y_unit = target_y_unit
df["unit"] = target_y_unit
for t in df.itertuples(index=False):
style = _make_line_args(props, t, df)
p.plot(t.vectime, t.vecvalue, label=legend_func(legend_cols, t, props), **style)
title = get_prop("title") or make_chart_title(df, title_cols)
set_plot_title(title)
ylabel = make_chart_title(df, ["title"])
if "xaxis_unit" in props or "yaxis_unit" in props:
_set_xlabel(p, props, x_unit, "Simulation Time")
_set_xlimits(p, props, x_unit)
_set_ylabel(p, props, y_unit, ylabel)
_set_ylimits(p, props, y_unit)
else:
# backward compatibility for old charts that don't have the xaxis_unit and yaxis_unit properties yet
p.xlabel("Simulation Time [s]")
if y_unit is not None:
ylabel += f" [{y_unit}]"
p.ylabel(ylabel)
if not ideplot.is_native_plot():
plt.gca().delta_measurement = delta_measurement.DeltaMeasurement(plt.gcf(), plt.gca())
[docs]
def plot_vectors_separate(df, props, legend_func=make_legend_label, sort=True):
"""
This is very similar to `plot_vectors`, with identical usage.
The only difference is in the end result, where each vector will
be plotted in its own separate set of axes (coordinate system),
arranged vertically, with a shared X axis during navigation.
"""
def get_prop(k):
return props[k] if k in props else None
title_cols, legend_cols = extract_label_columns(df, props)
if sort:
df.sort_values(by=legend_cols, inplace=True)
x_unit = "s"
# X unit
target_x_unit = get_prop("xaxis_unit")
if target_x_unit is None:
target_x_unit = x_unit
elif target_x_unit == "" and x_unit:
target_x_unit = _get_best_unit(df.vectime, x_unit)
if x_unit != target_x_unit and x_unit:
df.vectime = _convert_to_unit(df.vectime, x_unit, target_x_unit)
x_unit = target_x_unit
# compute endtime as the maximum timestamp in all vectors, or the upper limit of X axis
# TODO take the simulation duration instead, or the maximum of vector close times when they become available
endtimes = [t.vectime[-1] for t in df.itertuples(index=False) if t.vectime.size > 0]
endtime = np.max(endtimes) if endtimes else math.nan
if props.get("xaxis_max"):
endtime = max(endtime, _get_quantity(props, "xaxis_max", x_unit))
fig = plt.gcf()
ax = None
for i, t in enumerate(df.itertuples(index=False)):
style = _make_line_args(props, t, df)
ax = plt.subplot(df.shape[0], 1, i+1, sharex=ax)
if i != df.shape[0]-1:
plt.setp(ax.get_xticklabels(), visible=False)
ax.xaxis.get_label().set_visible(False)
if hasattr(t, "enum") and isinstance(t.enum, str) and t.enum and props.get("enum_as_strip") == "true":
_plot_enum(t.vectime, t.vecvalue, endtime, _parse_enum_spec(t.enum, True), label=legend_func(legend_cols, t, props), draw_edges=get_prop("enum_strip_edges") == "true")
ax.delta_measurement = delta_measurement.DeltaMeasurement(fig, ax, t.vectime)
else:
plt.plot(t.vectime, t.vecvalue, label=legend_func(legend_cols, t, props), **style)
ax.delta_measurement = delta_measurement.DeltaMeasurement(fig, ax)
# " means inches, % means percentage of total size
# horizontal left/right space around the whole plot: 0.08"+0%
# vertical top/bottom space around the whole plot: 0.01"+1%
# horizontal space between subplots: 0"+0%
# vertical space between subplots: 0.01"+5% divided between the subplots
# increasing h_pad from 0.01" would unnecessarily increase the distance between subplots
fig.set_constrained_layout_pads(w_pad=0.08, h_pad=0.01, wspace=0, hspace=0.05, rect=(0, 0.01, 1, 0.98))
plt.subplot(df.shape[0], 1, 1)
title = get_prop("title") or make_chart_title(df, title_cols)
set_plot_title(title)
if "xaxis_unit" in props or "yaxis_unit" in props:
plt.axes(ax)
_set_xlabel(plt, props, x_unit, "Simulation Time")
_set_xlimits(plt, props, x_unit)
else:
# backward compatibility for old charts that don't have the xaxis_unit property yet
plt.xlabel("Simulation Time [s]")
def _parse_enum_spec(enum_spec, reverse_mapping=False):
# "A, B, C" -> {'A': 0, 'B': 1, 'C': 2}
# "A=1, B, C=5, D" -> {'A': 1, 'B': 2, 'C': 5, 'D': 6}
kv_pairs = []
last_value = -1 # Start with -1 so the first default value will be 0
entries = enum_spec.split(',')
for entry in entries:
entry = entry.strip() # Remove any leading/trailing whitespace
if '=' in entry:
# Split on '=' and parse the key and value
key, value = entry.split('=')
value = int(value.strip())
kv_pairs.append((key.strip(), value))
last_value = value # Update last_value to the current explicit value
else:
# Increment last_value and use it for the current entry
last_value += 1
kv_pairs.append((entry, last_value))
if reverse_mapping:
return dict((v, k) for k, v in kv_pairs) # note: creating a normal dict THEN reversing it would be lossy if there are duplicates among the names
else:
return dict(kv_pairs)
def _plot_enum(vectime, vecvalue, endtime, labels_map, label, draw_edges):
label_colors = {index: "C" + str(index) for index, _ in enumerate(labels_map)}
colors = [label_colors[key] for key in sorted(label_colors.keys())]
boundaries = list(label_colors.keys()) + [max(label_colors.keys()) + 1]
cmap = ListedColormap(colors)
norm = BoundaryNorm(boundaries, cmap.N)
fig = plt.gcf()
ax = plt.gca()
ax.grid(alpha=0)
ax.tick_params(axis='both', which='both', labelleft=False, bottom=True, left=False)
# HACK: Adding this member causes the built-in "zoom" tool to leave the Y axis unaffected.
# See: https://github.com/matplotlib/matplotlib/blob/v3.10.1/lib/matplotlib/backend_bases.py#L3107-L3140
from dataclasses import dataclass
@dataclass
class FakeCbar:
orientation: str = "horizontal"
ax._colorbar = FakeCbar()
# ----
# HACK: This causes the built-in "pan" tool to leave the Y axis unaffected.
# See: https://github.com/matplotlib/matplotlib/blob/v3.10.1/lib/matplotlib/colorbar.py#L415-L419
# and https://github.com/matplotlib/matplotlib/blob/v3.10.1/lib/matplotlib/colorbar.py#L1325-L1332
# and https://github.com/matplotlib/matplotlib/blob/v3.10.1/lib/matplotlib/axes/_base.py#L4415-L4435
def drag_pan(ax, button, key, x, y):
points = ax._get_pan_points(button, key, x, y)
if points is not None:
ax.set_xlim(points[:, 0])
ax.set_ylim(0, 1)
import types
ax.drag_pan = types.MethodType(drag_pan, ax)
# ----
legend_handles, legend_labels = ax.get_legend_handles_labels()
legend_handles.append(plt.Rectangle((0, 0), 1, 1, color=(0.5, 0.5, 0.5, 0.5)))
legend_labels.append(label)
legend = ax.legend(legend_handles, legend_labels, markerscale = 0, handlelength = 0) # hide marker
legend.set_picker(True)
ax.default_legend = True
if vectime.size == 0:
return
endtime = max(endtime, vectime[-1])
vectime = np.append(vectime, endtime)
if draw_edges:
cex = ax.pcolormesh(vectime, [0, 1], np.array([vecvalue]), cmap=cmap, norm=norm, shading='flat', edgecolors=(0, 0, 0, .1), linewidth=0.1, mouseover=False)
else:
ax.pcolorfast(vectime, [0, 1], np.array([vecvalue]), cmap=cmap, norm=norm, mouseover=False)
def compute_conversion_factor(ax):
return ax.transData.get_matrix()[0, 0]
def on_xlim_changed(ax):
if ax is None:
return
for txt in ax.texts:
txt.remove()
x_min, x_max = ax.get_xlim()
if draw_edges:
c = np.sum((x_min < vectime) & (vectime < x_max))
cex.set_edgecolor((0, 0, 0, .5 * .99 ** c))
# Find start and end indices
start_idx = np.searchsorted(vectime, x_min) - 1
end_idx = np.searchsorted(vectime, x_max)
start_idx = max(start_idx, 0)
end_idx = min(end_idx, len(vectime)-1)
visible_widths = np.diff(vectime[start_idx:end_idx+1]) * compute_conversion_factor(ax)
textwidths = {}
for i in range(start_idx, end_idx):
width = visible_widths[i - start_idx]
if width > 10:
x = (vectime[i] + vectime[i+1]) / 2
value = vecvalue[i]
label = labels_map.get(value,value)
txt = ax.text(x, 0.5, label, fontsize='x-small', color='white', ha='center', va='center', clip_on=True)
if label in textwidths:
textwidth = textwidths[label]
else:
textwidth = txt.get_window_extent(fig.canvas.get_renderer()).width
textwidths[label] = textwidth
txt.set_rotation('horizontal' if textwidth < width else 'vertical')
def on_legend_click(ax):
ax.default_legend = not ax.default_legend
if ax.default_legend:
legend = ax.legend(legend_handles, legend_labels)
legend.set_picker(True)
else:
legend_elements = [plt.Rectangle((0, 0), 1, 1, color=color) for color in label_colors.values()]
legend = ax.legend(legend_elements, labels_map.values(), fontsize='x-small', loc='upper right', ncol=math.ceil(len(label_colors) / 4))
legend.set_picker(True)
legend.set_clip_on(False)
fig.canvas.draw_idle()
def on_resize_event(event):
for ax in fig.axes:
ax.callbacks.process('xlim_changed', ax)
def on_pick(event):
for ax in fig.get_axes():
if event.artist == ax.get_legend():
ax.callbacks.process('zlim_changed', ax) # KLUDGE: for switching between the two legends
ax.callbacks.connect('xlim_changed', on_xlim_changed)
ax.callbacks.connect('zlim_changed', on_legend_click) # KLUDGE: for switching between the two legends
fig.canvas.mpl_connect('resize_event', on_resize_event)
fig.canvas.mpl_connect('pick_event', on_pick)
[docs]
def plot_histograms(df, props, legend_func=make_legend_label, sort=True):
"""
Creates a histogram plot from the dataframe, with styling and additional input
coming from the properties. Each row in the dataframe defines a histogram.
Colors are assigned automatically. The `cycle_seed` property allows you to
select other combinations if the default one is not suitable.
A function to produce the legend labels can be passed in. By default,
`make_legend_label()` is used, which offers many ways to influence the
legend via dataframe columns and chart properties. In the absence of
more specified settings, the legend is normally computed from columns that best
differentiate among the histograms.
Parameters:
- `df`: The dataframe.
- `props` (dict): The properties.
- `legend_func` (function): The function to produce custom legend labels.
See `utils.make_legend_label()` for the prototype and semantics.
- `sort` (bool): Whether to sort the histograms by the columns used for the legend
(before applying `legend_func`, for backward bug-compatibility).
Columns of the dataframe:
- `binedges`, `binvalues` (array-like, `len(binedges)==len(binvalues)+1`):
The bin edges and the bin values (count or sum of weights) for the histogram.
- `min`, `max`, `underflows`, `overflows` (float, optional): The minimum/maximum
values, and the bin values for the underflow/overflow bins. These four columns
must either be all present or all absent from the dataframe.
- `legend` (string, optional): Legend label for the series. If missing,
legend labels are derived from other columns.
- `name`, `title`, `module`, etc. (optional): Provide input for the legend.
- `unit` (string, optional): The unit for the X-axis values. If present, it is required
to be the same for all series.
Notable properties that affect the plot:
- `normalize` (bool): If true, normalize the sum of the bin values to 1. If
`normalize` is true (and `cumulative` is false), the probability density
function (PDF) will be displayed.
- `cumulative` (bool): If true, show each bin as the sum of the previous bin
values plus itself. If both `normalize` and `cumulative` are true, that
results in the cumulative density function (CDF) being displayed.
- `show_overflows` (bool): If true, show the underflow/overflow bins.
- `title`: Plot title (autocomputed if missing).
- `drawstyle`: Selects whether to fill the area below the histogram line.
- `linestyle`, `linecolor`, `linewidth`: Styling.
- `cycle_seed`: Alters the sequence in which colors and markers are assigned to series.
- `xaxis_unit`: If present, specifies the unit for the X-axis. If empty string, automatically selects the best unit.
- `yaxis_unit`: If present, specifies the unit for the Y-axis. If empty string, automatically selects the best unit.
"""
x_unit = _check_same_unit(df)
y_unit = props.get("yaxis_unit")
p = ideplot if chart.is_native_chart() else plt
has_overflow_columns = "min" in df and "max" in df and "underflows" in df and "overflows" in df
if not has_overflow_columns:
if "min" in df or "max" in df or "underflows" in df or "overflows" in df:
raise ValueError("Either all or none of the following columns must be present: min, max, underflows, overflows")
def get_prop(k):
return props[k] if k in props else None
title_cols, legend_cols = extract_label_columns(df, props)
if sort:
df.sort_values(by=legend_cols, inplace=True)
# X unit
target_x_unit = get_prop("xaxis_unit")
if target_x_unit is None:
target_x_unit = x_unit
elif target_x_unit == "" and x_unit:
target_x_unit = _get_best_unit(df.binedges, x_unit)
if x_unit != target_x_unit and x_unit:
df.binedges = _convert_to_unit(df.binedges, x_unit, target_x_unit)
x_unit = target_x_unit
for t in df.itertuples(index=False):
style = _make_histline_args(props, t, df)
# We are branching here, and call the two .hist implementations
# differently, because the MPL API does not have support for
# underflows/overflows, so we have to "fake" them into the real
# data as additional cells. Drawing them separately wouldn't
# work, because that would change the histogram when the
# "normalized" or "cumulative" transforms are done (by MPL).
if chart.is_native_chart():
overflow = dict(minvalue=t.min, maxvalue=t.max, underflows=t.underflows, overflows=t.overflows) if has_overflow_columns else dict()
p.hist(t.binedges[:-1], t.binedges, weights=t.binvalues, label=legend_func(legend_cols, t, props), **overflow, **style)
else:
edges = list(t.binedges)
values = list(t.binvalues)
if has_overflow_columns:
has_underflow_bin = not np.isnan(t.min) and not np.isnan(t.underflows) and t.min < edges[0]
has_overflow_bin = not np.isnan(t.max) and not np.isnan(t.overflows) and t.max >= edges[-1]
if has_underflow_bin:
edges = [t.min] + edges
values = [t.underflows] + values
if has_overflow_bin:
edges = edges + [t.max]
values = values + [t.overflows]
p.hist(edges[:-1], edges, weights=values, label=legend_func(legend_cols, t, props), **style)
show_overflows = get_prop("show_overflows")
if show_overflows and chart.is_native_chart():
ideplot.set_property("Hist.ShowOverflowCell", str(_parse_optional_bool(show_overflows)).lower())
title = get_prop("title") or make_chart_title(df, title_cols)
set_plot_title(title)
if "xaxis_unit" in props or "yaxis_unit" in props:
_set_xlabel(p, props, x_unit, "")
_set_xlimits(p, props, x_unit)
_set_ylabel(p, props, y_unit, "")
_set_ylimits(p, props, y_unit)
else:
# backward compatibility for old charts that don't have the xaxis_unit property yet
if x_unit is not None:
p.xlabel(f"[{x_unit}]")
[docs]
def plot_lines(df, props, legend_func=make_legend_label, sort=True):
"""
Creates a line plot from the dataframe, with styling and additional input
coming from the properties. Each row in the dataframe defines a line.
Colors are assigned automatically. The `cycle_seed` property allows you to
select other combinations if the default one is not suitable.
A function to produce the legend labels can be passed in. By default,
`make_legend_label()` is used, which offers many ways to influence the
legend via dataframe columns and chart properties. In the absence of
more specified settings, the legend is normally computed from columns that best
differentiate among the lines.
Parameters:
- `df`: The dataframe.
- `props` (dict): The properties.
- `legend_func` (function): The function to produce custom legend labels.
See `utils.make_legend_label()` for the prototype and semantics.
- `sort` (bool): Whether to sort the series by the columns used for the legend
(before applying `legend_func`, for backward bug-compatibility).
Columns of the dataframe:
- `x`, `y` (array-like, `len(x)==len(y)`): The X and Y coordinates of the points.
- `error` (array-like, `len(x)==len(y)`, optional):
The half lengths of the error bars for each point.
- `legend` (string, optional): Legend label for the series. If missing,
legend labels are derived from other columns.
- `name`, `title`, `module`, etc. (optional): Provide input for the legend.
Notable properties that affect the plot:
- `title`: Plot title (autocomputed if missing).
- `linewidth`: Line width.
- `marker`: Marker style.
- `linestyle`, `linecolor`, `linewidth`: Styling.
- `error_style`: If `error` is present, controls how the error is shown.
Accepted values: "Error bars", "Error band"
- `cycle_seed`: Alters the sequence in which colors and markers are assigned to series.
- `unit`: If present, it is required to be the same for all series and will be used in the automatic y-axis label.
"""
unit = _check_same_unit(df)
p = ideplot if chart.is_native_chart() else plt
def get_prop(k):
return props[k] if k in props else None
title_cols, legend_cols = extract_label_columns(df, props)
if sort:
df.sort_values(by=legend_cols, inplace=True)
for t in df.itertuples(index=False):
style = _make_line_args(props, t, df)
if len(t.x) < 2 and style["marker"] == ' ':
style["marker"] = '.'
p.plot(t.x, t.y, label=legend_func(legend_cols, t, props), **style)
if hasattr(t, "error") and not ideplot.is_native_plot():
style["linewidth"] = float(style["linewidth"])
style["linestyle"] = "none"
if props["error_style"] == "Error bars":
plt.errorbar(t.x, t.y, yerr=t.error, capsize=float(props["cap_size"]), **style)
elif props["error_style"] == "Error band":
plt.fill_between(t.x, t.y-t.error, t.y+t.error, alpha=float(props["band_alpha"]))
title = get_prop("title") or make_chart_title(df, title_cols)
set_plot_title(title)
if "xaxis_unit" in props or "yaxis_unit" in props:
_set_ylabel(p, props, unit, "")
else:
# backward compatibility for old charts that don't have the yaxis_unit property yet
if unit is not None:
p.ylabel(f"[{unit}]")
[docs]
def plot_boxwhiskers(df, props, legend_func=make_legend_label, sort=True):
"""
Creates a box and whiskers plot from the dataframe, with styling and additional
input coming from the properties. Each row in the dataframe defines one set
of a box and two whiskers.
Colors are assigned automatically. The `cycle_seed` property allows you to
select other combinations if the default one is not suitable.
A function to produce the legend labels can be passed in. By default,
`make_legend_label()` is used, which offers many ways to influence the
legend via dataframe columns and chart properties. In the absence of
more specified settings, the legend is normally computed from columns that best
differentiate among the boxes.
Parameters:
- `df`: The dataframe.
- `props` (dict): The properties.
- `legend_func` (function): The function to produce custom legend labels.
See `utils.make_legend_label()` for the prototype and semantics.
- `sort` (bool): Whether to sort the series by the columns used for the legend
(before applying `legend_func`, for backward bug-compatibility).
Columns of the dataframe:
- `min`, `max`, `mean`, `stddev`, `count` (float): The minimum/maximum
values, mean, standard deviation, and sample count of the data.
- `legend` (string, optional): Legend label for the series. If missing,
legend labels are derived from other columns.
- `name`, `title`, `module`, etc. (optional): Provide input for the legend.
Notable properties that affect the plot:
- `title`: Plot title (autocomputed if missing).
- `cycle_seed`: Alters the sequence in which colors and markers are assigned to series.
- `unit`: If present, it is required to be the same for all series and will be used in the automatic y-axis label.
"""
y_unit = _check_same_unit(df)
title_cols, legend_cols = extract_label_columns(df, props)
def get_prop(k):
return props[k] if k in props else None
if sort:
df.sort_values(by=legend_cols, axis='index', inplace=True)
# Y unit
target_y_unit = get_prop("yaxis_unit")
if target_y_unit is None:
target_y_unit = y_unit
elif target_y_unit == "" and y_unit:
target_y_unit = _get_best_unit(df["mean"], y_unit)
if y_unit != target_y_unit and y_unit:
df["min"] = _convert_to_unit(df["min"], y_unit, target_y_unit)
df["mean"] = _convert_to_unit(df["mean"], y_unit, target_y_unit)
df["stddev"] = _convert_to_unit(df["stddev"], y_unit, target_y_unit)
df["max"] = _convert_to_unit(df["max"], y_unit, target_y_unit)
y_unit = target_y_unit
# This is how much of the standard deviation will give the 25th and 75th
# percentiles, assuming normal distribution.
# >>> math.sqrt(2) * scipy.special.erfinv(0.5)
coeff = 0.6744897501960817
boxes = [(r.min, r.mean - r.stddev * coeff, r.mean, r.mean + r.stddev * coeff, r.max)
for r in df.itertuples(index=False) if r.count > 0]
labels = [legend_func(legend_cols, r, props) for r in df.itertuples(index=False) if r.count > 0]
customized_box_plot(boxes, labels)
title = make_chart_title(df, title_cols)
if "title" in props and props["title"]:
title = props["title"]
set_plot_title(title)
if "yaxis_unit" in props:
_set_ylabel(plt, props, y_unit, "")
_set_ylimits(plt, props, y_unit)
else:
# backward compatibility for old charts that don't have the xaxis_unit property yet
if y_unit is not None:
plt.ylabel(f"[{y_unit}]")
# source: https://stackoverflow.com/a/39789718/635587
[docs]
def customized_box_plot(percentiles, labels=None, axes=None, redraw=True, *args, **kwargs):
"""
Generates a customized box-and-whiskers plot based on explicitly specified
percentile values. This method is necessary because `pyplot.boxplot()` insists
on computing the stats from the raw data (which we often don't have) itself.
The data is in the `percentiles` argument, which should be a list of tuples.
One box will be drawn for each tuple. Each tuple contains 6 elements (or 5,
because the last one is optional):
(*q1_start*, *q2_start*, *q3_start*, *q4_start*, *q4_end*, *fliers*)
The first five elements have the following meaning:
- *q1_start*: y coord of bottom whisker cap
- *q2_start*: y coord of bottom of the box
- *q3_start*: y coord of median mark
- *q4_start*: y coord of top of the box
- *q4_end*: y coord of top whisker cap
The last element, *fliers*, is a list containing the values of the
outlier points.
x coords of the box-and-whiskers plots are automatic.
Parameters:
- `percentiles`: The list of tuples.
- `labels`: If provided, the legend labels for the boxes.
- `axes`: The axes object of the plot.
- `redraw`: If False, redraw is deferred.
- `args`, `kwargs`: Passed to `axes.boxplot()`.
"""
if axes is None:
axes = plt.gca()
min_y, max_y = float('inf'), -float('inf')
if labels is not None:
if len(labels) != len(percentiles):
raise ValueError("There must be as many labels as elements in the percentiles list")
for box_no, pdata in enumerate(percentiles):
color = next(_color_cycle)
box_plot = axes.boxplot([-9, -4, 2, 4, 9], positions=[box_no], widths=[0.5], showmeans=True,
meanprops=dict(marker='+', markeredgecolor=mpl.rcParams["axes.facecolor"]),
medianprops=dict(linestyle=None, linewidth=0), # hide the median line
boxprops=dict(facecolor=color),
whiskerprops=dict(zorder=5, linewidth=1, solid_capstyle="butt"),
capprops=dict(zorder=6, linewidth=2, color=color, solid_capstyle="butt"),
patch_artist=True, *args, **kwargs)
if labels is not None:
# boxplot does not register its patches as legend handles, as it only wants to
# put the labels on axis ticks.
# And we also don't want to explicitly specify the legend contents, because the
# legend() call is sometime later, on postconfigure_plot. Only one thing remains:
# To make a "fake" bar for each box, and remove it - but not entirely (as that)
# would also remove it from the Legend... (same with hiding) - so its single
# patch is removed instead...
bar = axes.bar([0], [0], color=color, label=labels[box_no])
bar.patches[0].remove()
if len(pdata) == 6:
(q1_start, q2_start, q3_start, q4_start, q4_end, fliers) = pdata
elif len(pdata) == 5:
(q1_start, q2_start, q3_start, q4_start, q4_end) = pdata
fliers = []
else:
raise ValueError("Percentile arrays for customized_box_plot must have either 5 or 6 values")
fliers = np.array(fliers)
# Lower cap
box_plot['caps'][0].set_ydata([q1_start, q1_start])
# xdata is determined by the width of the box plot
# Lower whiskers
box_plot['whiskers'][0].set_ydata([q1_start, q2_start])
# Higher cap
box_plot['caps'][1].set_ydata([q4_end, q4_end])
# Higher whiskers
box_plot['whiskers'][1].set_ydata([q4_start, q4_end])
# Box
path = box_plot['boxes'][0].get_path()
path.vertices[0][1] = q2_start
path.vertices[1][1] = q2_start
path.vertices[2][1] = q4_start
path.vertices[3][1] = q4_start
path.vertices[4][1] = q2_start
box_plot['means'][0].set_ydata([q3_start, q3_start])
box_plot['medians'][0].set_visible(False)
# Outliers
if len(fliers) > 0:
# If outliers exist
box_plot['fliers'][0].set(xdata = [box_no] * len(fliers),
ydata = fliers)
min_y = min(q1_start, min_y, fliers.min())
max_y = max(q4_end, max_y, fliers.max())
else:
min_y = min(q1_start, min_y)
max_y = max(q4_end, max_y)
mid_y = (min_y + max_y) / 2
# The y axis is rescaled to fit the new box plot completely with 10%
# of the maximum value at both ends
axes.set_ylim([mid_y - (mid_y - min_y) * 1.25, mid_y + (max_y - mid_y) * 1.25])
axes.set_xlim(-0.5, len(percentiles)-0.5)
axes.set_xticks([])
axes.set_xticklabels([])
# If redraw is set to true, the canvas is updated.
if redraw:
axes.figure.canvas.draw_idle()
def _get_quantity(props, k, target_unit, default_unit=None):
return _parse_quantity(props.get(k), target_unit, default_unit)
def _parse_quantity(v, target_unit, default_unit=None):
if not target_unit:
return float(v)
try:
fv = float(v)
if not default_unit:
default_unit = uc.getBaseUnit(target_unit)
if not default_unit:
return fv
v += default_unit
except (ValueError, TypeError):
pass # most likely, v already has a unit appended
return uc.parseQuantity(v, target_unit)
def _set_xlimits(p, props, target_unit):
if props.get("xaxis_min"):
p.xlim(left=_get_quantity(props, "xaxis_min", target_unit))
if props.get("xaxis_max"):
p.xlim(right=_get_quantity(props, "xaxis_max", target_unit))
def _set_ylimits(p, props, target_unit):
if props.get("yaxis_min"):
p.ylim(bottom=_get_quantity(props, "yaxis_min", target_unit))
if props.get("yaxis_max"):
p.ylim(top=_get_quantity(props, "yaxis_max", target_unit))
def _set_xlabel(p, props, unit, label):
if props.get("xaxis_title"):
label = props.get("xaxis_title")
if props.get("xaxis_unit") is not None and unit is not None and len(unit) > 0:
if len(label) > 0:
label += " "
label += f"[{unit}]"
p.xlabel(label)
def _set_ylabel(p, props, unit, label):
if props.get("yaxis_title"):
label = props.get("yaxis_title")
if props.get("yaxis_unit") is not None and unit is not None and len(unit) > 0:
if len(label) > 0:
label += " "
label += f"[{unit}]"
p.ylabel(label)
def _legend_loc_outside_args(loc):
mapping = {
"outside top left": ("lower left", (0, 1.05)),
"outside top center": ("lower center", (0.5, 1.05)),
"outside top right": ("lower right", (1, 1.05)),
"outside bottom left": ("upper left", (0, -0.05)),
"outside bottom center": ("upper center", (0.5, -0.05)),
"outside bottom right": ("upper right", (1, -0.05)),
"outside left top": ("upper right", (-0.03, 1)),
"outside left center": ("center right", (-0.03, 0.5)),
"outside left bottom": ("lower right", (-0.03, 0)),
"outside right top": ("upper left", (1.03, 1)),
"outside right center": ("center left", (1.03, 0.5)),
"outside right bottom": ("lower left", (1.03, 0)),
}
if loc not in mapping:
raise ValueError("loc='{}' is not recognized/supported".format(loc))
(anchorloc, relpos) = mapping[loc]
return {"loc": anchorloc, "bbox_to_anchor": relpos}
[docs]
def postconfigure_plot(props):
"""
Configures the plot according to the given properties, which normally
get their values from settings in the "Configure Chart" dialog.
Calling this function after plotting is performed should be a standard part
of chart scripts.
A partial list of properties taken into account:
- `yaxis_title`, `yaxis_title`, `xaxis_min`, `xaxis_max`, `yaxis_min`,
`yaxis_max`, `xaxis_log`, `yaxis_log`, `legend_show`, `legend_border`,
`legend_placement`, `grid_show`, `grid_density`
- properties listed in the `plot.properties` property
Parameters:
- `props` (dict): the properties
"""
p = ideplot if ideplot.is_native_plot() else plt
def get_prop(k):
return props[k] if k in props else None
def setup():
# Backward compatibility: If the chart has no unit properties,
# it was created before unit conversion support was added, so
# keep the old behavior of setting (potentially overwriting)
# the axis labels/limits here.
if "xaxis_unit" not in props and "yaxis_unit" not in props:
if get_prop("xaxis_title"):
p.xlabel(get_prop("xaxis_title"))
if get_prop("yaxis_title"):
p.ylabel(get_prop("yaxis_title"))
if get_prop("xaxis_min"):
p.xlim(left=float(get_prop("xaxis_min")))
if get_prop("xaxis_max"):
p.xlim(right=float(get_prop("xaxis_max")))
if get_prop("yaxis_min"):
p.ylim(bottom=float(get_prop("yaxis_min")))
if get_prop("yaxis_max"):
p.ylim(top=float(get_prop("yaxis_max")))
if get_prop("xaxis_log"):
p.xscale("log" if _parse_optional_bool(get_prop("xaxis_log")) else "linear")
if get_prop("yaxis_log"):
p.yscale("log" if _parse_optional_bool(get_prop("yaxis_log")) else "linear")
ideplot.grid(_parse_optional_bool(get_prop("grid_show")),
"major" if (get_prop("grid_density") or "").lower() == "major" else "both") # grid_density is "Major" or "All"
if ideplot.is_native_plot():
setup()
ideplot.legend(show=_parse_optional_bool(get_prop("legend_show")),
frameon=_parse_optional_bool(get_prop("legend_border")),
loc=get_prop("legend_placement"))
ideplot.set_properties(parse_rcparams(get_prop("plot.properties") or ""))
else:
for ax in p.gcf().axes:
plt.sca(ax)
setup()
if _parse_optional_bool(get_prop("legend_show")):
loc = get_prop("legend_placement")
args = { "loc": loc}
if loc and loc.startswith("outside"):
args.update(_legend_loc_outside_args(loc))
args["frameon"] =_parse_optional_bool(get_prop("legend_border"))
handles, labels = ax.get_legend_handles_labels()
# if there is only one legend entry, make the symbol invisible -- this is especially useful for plot_vectors_separate()
if len(handles) <= 1:
args["markerscale"] = 0
args["handlelength"] = 0
if handles:
plt.legend(**args)
[docs]
def export_image_if_needed(props):
"""
If a certain property is set, save the plot in the selected image format.
Calling this function should be a standard part of chart scripts, as it is what
makes the "Export image" functionality of the IDE and `opp_charttool` work.
Note that for export, even IDE-native charts are rendered using Matplotlib.
The properties that are taken into account:
- `export_image` (boolean): Controls whether to perform the exporting. This is
normally `false`, and only set to `true` by the IDE or opp_charttool when
image export is requested.
- `image_export_format`: The default is SVG. Accepted formats (and their names) are the ones supported by Matplotlib.
- `image_export_folder`: The folder in which the image file is to be created.
- `image_export_filename`: The output file name. If it has no extension,
one will be added based on the format. If missing or empty, a sanitized
version of the chart name is used.
- `image_export_width`: Image width in inches (default: 6")
- `image_export_height`: Image height in inches (default: 4")
- `image_export_dpi`: DPI setting, default 300. For raster image formats, the
image dimensions are produced as width (or height) times dpi.
Note that these properties come from two sources to allow meaningful batch
export. `export_image`, `image_export_format`, `image_export_folder` and
`image_export_dpi` come from the export dialog because they are common
to all charts, while `image_export_filename`, `image_export_width` and
`image_export_height` come from the chart properties because they are
specific to each chart. Note that `image_export_dpi` is used for controlling
the resolution (for raster image formats) while letting charts maintain
their own aspect ratio and relative sizes.
Parameters:
- `props` (dict): the properties
"""
def get_prop(k):
return props[k] if k in props else None
if _parse_optional_bool(get_prop("export_image")):
filepath = get_image_export_filepath(props)
format = get_prop("image_export_format") or "svg"
width = float(get_prop("image_export_width") or 6)
height = float(get_prop("image_export_height") or 4)
dpi = float(get_prop("image_export_dpi") or "300")
logger.info(f"exporting image to: '{filepath}' as {format}")
os.makedirs(os.path.dirname(filepath) or ".", exist_ok=True)
plt.gcf().set_size_inches(width, height)
plt.gcf().set_dpi(dpi)
# in case of Matplotlib-emulated native widgets, make it look more similar to the IDE
if chart.is_native_chart():
ax = plt.gca()
# add digit grouping to the axis tick labels
ax.xaxis.set_minor_formatter(_DigitGroupingFormatter())
ax.xaxis.set_major_formatter(_DigitGroupingFormatter())
ax.yaxis.set_minor_formatter(_DigitGroupingFormatter())
ax.yaxis.set_major_formatter(_DigitGroupingFormatter())
# only use scientific notation for negative exponents
ax.ticklabel_format(scilimits=(0, 1000), useOffset=False, useLocale=False)
from enum import Enum
class PlotAreaCalculationMode(Enum):
DATA_WITH_PADDING = 0
DATA_WITH_PADDING_AND_ORIGIN = 1
DATA_WITH_PADDING_AND_ORIGIN_WITH_PADDING = 2
mode = PlotAreaCalculationMode.DATA_WITH_PADDING_AND_ORIGIN
# get original dimensions
original_min_x, original_max_x = ax.get_xlim()
original_min_y, original_max_y = ax.get_ylim()
# start with a tight value bounding box
ax.margins(0)
ax.autoscale(True)
# apply the same auto-range logic as in the native widgets of the IDE
min_x, max_x = ax.get_xlim()
min_y, max_y = ax.get_ylim()
# relative to data range
padding_left = 0.01
padding_right = 0.01
padding_bottom = 0.1
padding_top = 0.1
if mode in (PlotAreaCalculationMode.DATA_WITH_PADDING_AND_ORIGIN_WITH_PADDING,
PlotAreaCalculationMode.DATA_WITH_PADDING_AND_ORIGIN):
# The idea is that if we're extending the axis range just to include
# the origin, we probably don't want as much padding around it.
origin_margin_factor = 0.0 if (mode == PlotAreaCalculationMode.DATA_WITH_PADDING_AND_ORIGIN) \
else 0.5
if min_x >= 0:
min_x = 0
padding_left *= origin_margin_factor
if max_x <= 0:
max_x = 0
padding_right *= origin_margin_factor
if min_y >= 0:
min_y = 0
padding_bottom *= origin_margin_factor
if max_y <= 0:
max_y = 0
padding_top *= origin_margin_factor
w = max_x - min_x
h = max_y - min_y
if w == 0:
min_x -= 0.5
max_x += 0.5
else:
min_x -= w * padding_left
max_x += w * padding_right
if h == 0:
min_y -= 0.5
max_y += 0.5
else:
min_y -= h * padding_bottom
max_y += h * padding_top
min_x = float(original_min_x if get_prop("xaxis_min") else min_x)
max_x = float(original_max_x if get_prop("xaxis_max") else max_x)
min_y = float(original_min_y if get_prop("yaxis_min") else min_y)
max_y = float(original_max_y if get_prop("yaxis_max") else max_y)
ax.set_xlim(left=min_x, right=max_x)
ax.set_ylim(bottom=min_y, top=max_y)
def _make_image_metadata():
# Include versions of various software components into the metadata,
# to facilitate resolving support reqests later. We use "Creator" as key,
# as it seems to be suitable with all image formats. Some formats ignore it
# and PNG would prefer to call it "Software", but it doesn't cause an exception
# to be thrown from savefig(). JPG and TIF display a DeprecationWarning though:
# "savefig() got unexpected argument 'metadata' which is no longer supported
# as of 3.3 and will become an error two minor releases later"
import platform
creatorInfo = f"omnetpp.scave {__version__}; Matplotlib {mpl.__version__} w/ backend {mpl.get_backend()}; Pandas {pd.__version__}; Python {platform.python_version()}; {platform.platform()}"
return { "Creator" : creatorInfo }
# Explicitly select a non-interactive backend appropriate for the export format,
# and keep auto-selection for any other formats not listed here.
backend = _get_mpl_backend_for_image_format(format)
# save image
try:
retry = False
plt.savefig(filepath, format=format, dpi=dpi, metadata=_make_image_metadata(), backend=backend)
except:
retry = True
if retry:
# try again without metadata, in case that caused the error
# oh yes, and we cannot put this into the "except" block, because we'd get
# an ugly error message ("During handling of the above exception, another
# exception occurred") if this fails too, e.g. due to an invalid "format" parameter
plt.savefig(filepath, format=format, dpi=dpi, backend=backend)
global verbose_export
if verbose_export:
print(f'Exported image to: "{filepath}" as {format}')
[docs]
def get_image_export_filepath(props):
"""
Returns the file path for the image to export based on the
`image_export_format`, `image_export_folder` and
`image_export_filename` properties given in `props`.
If a relative filename is returned, it is relative to the
working directory when the image export takes place.
"""
def get_prop(k):
return props[k] if k in props else None
format = get_prop("image_export_format") or "svg"
folder = get_prop("image_export_folder") or "."
filename = get_prop("image_export_filename") or _sanitize_filename(chart.get_name())
filename = filename + "." + format
return os.path.normpath(os.path.join(folder, filename))
def _format_to_extension(format):
format = format.lower()
format_to_extension = {
"markdown": "md",
"latex": "tex",
"excel": "xlsx",
"xls": "xlsx",
"bigquery": "gbq",
"stata": "dta",
"hdf": "h5",
"feather": "ftr",
"pickle": "pkl"
}
if format in format_to_extension.keys():
format = format_to_extension[format]
return format
def _get_mpl_backend_for_image_format(format):
return {
"png": "AGG",
"jpg": "AGG",
"jpeg": "AGG",
"tif": "AGG",
"tiff": "AGG",
"raw": "AGG",
"svg": "SVG",
"svgz": "SVG",
"ps" : "PS",
"eps": "PS",
"pdf": "PDF",
"pgf": "PGF",
}.get(format.lower(), None)
def _export_df_as(df, format, filepath, **kwargs):
extension = _format_to_extension(format)
if extension == "pkl":
df.to_pickle(filepath, **kwargs)
elif extension == "csv":
df.to_csv(filepath, **kwargs)
elif extension == "h5":
if "key" not in kwargs.keys():
kwargs["key"] = "results"
df.to_hdf(filepath, **kwargs)
#elif extension == "sql":
# df.to_sql(filepath, **kwargs) # needs a connection
elif extension == "xlsx":
df.to_excel(filepath, **kwargs)
elif extension == "json":
df.to_json(filepath, **kwargs)
elif extension == "html":
df.to_html(filepath, **kwargs)
#elif extension == "ftr":
# df.reset_index().to_feather(filepath, **kwargs) # column names must be strings
elif extension == "tex":
df.to_latex(filepath, **kwargs)
elif extension == "dta":
df.to_stata(filepath, **kwargs)
#elif extension == "gbq":
# df.to_gbq(filepath, **kwargs) # needs authentication
elif extension == "md":
df.to_markdown(filepath, **kwargs)
else:
raise ValueError("Unknown export data format: " + format)
[docs]
def export_data_if_needed(df, props, **kwargs):
"""
If a certain property is set, save the dataframe in CSV format.
Calling this function should be a standard part of chart scripts, as it is what
makes the "Export data" functionality of the IDE and `opp_charttool` work.
The properties that are taken into account:
- `export_data` (boolean): Controls whether to perform the exporting. This is
normally `false`, and only set to `true` by the IDE or opp_charttool when
data export is requested.
- `data_export_folder`: The folder in which the CSV file is to be created.
- `data_export_filename`: The output file name. If missing or empty, a
sanitized version of the chart name is used.
Note that these properties come from two sources to allow meaningful batch
export. `export_data` and `image_export_folder` come from the export dialog
because they are common to all charts, and `image_export_filename` comes
from the chart properties because it is specific to each chart.
Parameters:
- `df`: the dataframe to save
- `props` (dict): the properties
"""
try:
chart.set_observed_column_names(list(df.columns.values))
except Exception as e:
logger.warning(f"error in chart.set_observed_column_names(): {e}")
def get_prop(k):
return props[k] if k in props else None
if _parse_optional_bool(get_prop("export_data")):
format = get_prop("data_export_format") or "csv"
filepath = get_data_export_filepath(props)
logger.info(f"exporting data to: '{filepath}' as {format}")
os.makedirs(os.path.dirname(filepath) or ".", exist_ok=True)
with np.printoptions(threshold=sys.maxsize, linewidth=sys.maxsize), \
pd.option_context('display.max_columns', None,
'display.max_colwidth', None):
_export_df_as(df, format, filepath, **kwargs)
global verbose_export
if verbose_export:
print(f'Exported data to: "{filepath}" as {format}')
[docs]
def get_data_export_filepath(props):
"""
Returns the file path for the data to export based on the
`data_export_format`, `data_export_folder` and
`data_export_filename` properties given in `props`.
If a relative filename is returned, it is relative to the
working directory when the data export takes place.
"""
def get_prop(k):
return props[k] if k in props else None
format = get_prop("data_export_format") or "csv"
extension = _format_to_extension(format)
folder = get_prop("data_export_folder") or "."
filename = get_prop("data_export_filename") or _sanitize_filename(chart.get_name())
filename = filename + "." + extension
return os.path.normpath(os.path.join(folder, filename))
def _sanitize_filename(filename):
valid_chars = "-_=.() %s%s" % (string.ascii_letters, string.digits) # TODO may be too strict, e.g. '#' or accented letters should be allowed
return ''.join(c for c in filename if c in valid_chars)
def _initialize_cycles(props):
"""
Initialize the `_marker_cycle` and `_color_cycle` global variables, taking
the `cycle_seed` property into account.
"""
def get_prop(k):
return props[k] if k in props else None
global _marker_cycle
global _color_cycle
seed = int(get_prop('cycle_seed') or 0)
r = random.Random(seed)
ml = list("osv^<>pDd")
if seed != 0:
r.shuffle(ml)
_marker_cycle = it.cycle(ml)
prop_cycler = plt.rcParams['axes.prop_cycle']
if chart.is_native_chart():
num_colors = 10
else:
num_colors = len(prop_cycler.by_key()['color']) if "color" in prop_cycler.keys else 1
cl = ["C" + str(i) for i in range(num_colors)]
if seed != 0:
r.shuffle(cl)
_color_cycle = it.cycle(cl)
[docs]
def histogram_bin_edges_from_props(values, unit, props):
"""
Calculates histogram bin edges based on chart dialog properties.
This is a convenience wrapper around `histogram_bin_edges()`. See that function
for detailed documentation of the binning algorithm and parameters.
Parameters:
- `values`: Input data array
- `unit`: Measurement unit of the input data
- `props` (dict): Properties from the chart configuration dialog, containing
settings like "use_manual", "manual_edges", "num_bins", "bin_width",
"range_min", "range_max", "round_bin_widths", "method".
Returns:
- bin_edges: Array of bin edges
"""
use_manual = props.get("use_manual") == "true"
if use_manual:
# Parse manual bin edges (space or comma-separated values)
manual_edges_str = props.get("manual_edges", "").strip()
manual_edges = None
if manual_edges_str:
edges_list = re.split(r'[,\s]+', manual_edges_str)
manual_edges = np.array([_parse_quantity(x.strip(), unit) for x in edges_list if x.strip()])
if manual_edges is None or len(manual_edges) < 2:
raise chart.ChartScriptError("Manual bin edges must have at least two values.")
return manual_edges
else:
# Extract constraint parameters from form, converting empty strings to None
def get_optional_int(key):
val = props.get(key, "").strip()
return int(val) if val else None
def get_optional_quantity(key):
val = props.get(key, "").strip()
return _get_quantity(props, key, unit) if val else None
constraint_params = {
'num_bins': get_optional_int("num_bins"),
'bin_width': get_optional_quantity("bin_width"),
'range_min': get_optional_quantity("range_min"),
'range_max': get_optional_quantity("range_max"),
'round_bin_widths': props.get("round_bin_widths", "true") == "true",
'method': props.get("method", "auto") if not get_optional_quantity("bin_width") else None
}
try:
bin_edges = histogram_bin_edges(values, **constraint_params)
return bin_edges
except ValueError as ex:
raise chart.ChartScriptError(ex.args[0])
[docs]
def histogram_bin_edges(values, num_bins=None, bin_width=None, range_min=None, range_max=None, method=None, round_bin_widths=True):
"""
Enhanced histogram bin edge calculator with flexible parameter combinations
and smart defaults. Automatically returns integer-aligned edges for integer
data and uses rounded "nice" bin widths for improved readability.
Parameters:
- values: Input data array
- num_bins: Number of bins to use. If round_bin_widths is True, this will be treated as a hint.
- bin_width: Fixed width for each bin
- range_min: Lower endpoint of histogram range
- range_max: Upper endpoint of histogram range
- round_bin_widths: Choose "nice" bin widths such as 1, 2, 5 * 10^k.
- method: Binning method (used if num_bins is None, default: 'auto' if None)
When parameters are provided (not None), they are respected.
If parameters are overconstrained, an exception will be thrown.
Returns:
- bin_edges: Array of bin edges
Raises:
- ValueError: If parameters are overconstrained or incompatible
"""
has_num_bins = num_bins is not None
has_bin_width = bin_width is not None
has_range_min = range_min is not None
has_range_max = range_max is not None
has_method = method is not None
# Basic validation of input array (empty/NaN/inf)
values = np.asarray(values)
if values.size == 0:
raise ValueError("values must be non-empty")
if not np.isfinite(values).all():
raise ValueError("values must be finite (no NaN/inf)")
# Validate parameter constraints
if has_bin_width and has_method:
raise ValueError("Cannot specify both 'bin_width' and 'method' at the same time")
if has_num_bins and not isinstance(num_bins, int):
raise ValueError("num_bins must be an integer")
if has_num_bins and num_bins <= 0:
raise ValueError(f"num_bins must be positive, got {num_bins}")
if has_bin_width and bin_width <= 0:
raise ValueError(f"bin_width must be positive, got {bin_width}")
# First, handle fully constrained range
if has_range_min and has_range_max:
range = range_max - range_min
if range <= 0:
raise ValueError(f"range_min < range_max required, got {range_min=} and {range_max=}")
values_range_covered = values.min() >= range_min and values.max() < range_max
if not values_range_covered:
logger.warning(f"Histogram range ({range_min},{range_max}) does not fully cover values range ({values.min()},{values.max()})")
if has_num_bins and not has_bin_width:
bin_width = range / num_bins
elif not has_num_bins and has_bin_width:
num_bins = round(range / bin_width)
bin_width = range / num_bins # adjust
elif not has_num_bins and not has_bin_width:
# use numpy to tell us the optimal number of bins
tmp_edges = np.histogram_bin_edges(values, method or "auto", (range_min, range_max))
num_bins = len(tmp_edges) - 1
bin_width = range / num_bins
else:
raise ValueError(f"Do not specify num_bins, bin_width, range_min, and range_max together")
bin_edges = range_min + np.arange(num_bins+1) * bin_width
bin_edges[-1] = range_max
assert range_min == bin_edges[0]
assert range_max == bin_edges[-1]
return bin_edges
def nearest_nice_number(num):
"""Return the next number of the form 1, 2, 2.5, 5, 10* 10^k"""
assert num > 0
magnitude = math.floor(math.log10(abs(num)))
scaled = num / (10 ** magnitude) # [1, 10)
# scaled = round(scaled*2)/2 if scaled < 3 else round(scaled)
scaled = 1 if scaled < 1.5 else 2 if scaled < 2.2 else 2.5 if scaled < 3.5 else 5 if scaled < 7 else 10
return scaled * (10 ** magnitude)
# Else, first compute bin_width
all_integers = None
if not has_bin_width:
# compute 1st approx
eff_min = range_min if has_range_min else values.min()
eff_max = range_max if has_range_max else values.max()
if not has_num_bins:
# use numpy to tell us the optimal number of bins
tmp_edges = np.histogram_bin_edges(values, method or "auto", (eff_min, eff_max))
num_bins = len(tmp_edges) - 1
bin_width = math.fabs(eff_max - eff_min) / num_bins
if bin_width == 0:
bin_width = 1
# adjust to make it a nice round number
all_integers = not np.mod(values, 1).any()
if all_integers:
bin_width = max(1, round(bin_width))
if round_bin_widths:
# note: the way we "cheat" here will affect the number of bins!
bin_width = nearest_nice_number(bin_width)
# Compute edges from bin_width and range constraints
# Note: We don't use num_bins to avoid the error of accidentally leaving values uncovered.
# Instead, bin_width must be chosen so that it produces the desired number of bins.
assert not has_range_min or not has_range_max
if has_range_min:
n = max(1, math.ceil((values.max() - range_min) / bin_width))
bin_edges = range_min + np.arange(n+1) * bin_width
if bin_edges[-1] <= values.max():
bin_edges = np.append(bin_edges, bin_edges[-1] + bin_width) # extend to include values.max())
elif has_range_max:
n = max(1, math.ceil((range_max - values.min()) / bin_width))
bin_edges = range_max - np.arange(n+1, -1, -1) * bin_width
else:
# cover values range with bin_width bins, make edges multiple of bin_width
i_min = math.floor(values.min() / bin_width)
i_max = math.ceil(values.max() / bin_width)
bin_edges = np.arange(i_min, i_max+1) * bin_width
if bin_edges[-1] <= values.max():
bin_edges = np.append(bin_edges, bin_edges[-1] + bin_width) # extend to include values.max())
# Issue warning if histogram range does not cover all values (note: bins' right edges are open, i.e. exclusive)
values_range_covered = values.min() >= bin_edges[0] and values.max() < bin_edges[-1]
if not values_range_covered:
logger.warning(f"Histogram range ({bin_edges[0]},{bin_edges[-1]}) does not fully cover values range ({values.min()},{values.max()})")
# Assert that constraints are met
assert not has_num_bins or round_bin_widths or all_integers or num_bins == len(bin_edges)-1 # num_bins is fulfilled *unless* this-or-that...
assert not has_bin_width or bin_width == bin_edges[1] - bin_edges[0]
assert not has_range_min or range_min == bin_edges[0]
assert not has_range_max or range_max == bin_edges[-1]
# histogram_width_constrained = has_range_min or has_range_max or (has_num_bins and has_bin_width)
# assert histogram_width_constrained or values_range_covered
return bin_edges
[docs]
def confidence_interval(alpha, data):
"""
Returns the half-length of the confidence interval of the mean of `data`, assuming
normal distribution, for the given confidence level `alpha`.
Parameters:
- `alpha` (float): Confidence level, must be in the [0..1] range.
- `data` (array-like): An array containing the values.
"""
return math.nan if len(data) <= 1 else 0 if len(set(data)) <= 1 else st.norm.interval(alpha, loc=0, scale=st.sem(data))[1]
[docs]
def pivot_for_barchart(df, groups, series, confidence_level=None, sort=True):
"""
Turns a DataFrame containing scalar results (in the format returned
by `results.get_scalars()`) into a 3-tuple of a value, an error, and
a metadata DataFrame, which can then be passed to `utils.plot_bars()`.
The error dataframe is None if no confidence level is given.
Parameters:
- `df` (pandas.DataFrame): The dataframe to pivot.
- `groups` (list): A list of column names, the values in which will
be used as names for the bar groups.
- `series` (list): A list of column names, the values in which will
be used as names for the bar series.
- `confidence_level` (float, optional):
The confidence level to use when computing the sizes of the error bars.
- `sort` (bool): Whether to sort the values by the columns in `groups`
and `series` before pivoting.
Returns:
- A triplet of DataFrames containing the pivoted data: (values, errors, metadata)
"""
df = to_numeric(df, groups + series)
if sort:
df.sort_values(by=groups+series, axis='index', inplace=True)
def aggfunc(values):
if values.empty:
return None
elif pd.api.types.is_numeric_dtype(values):
return values.mean()
else:
uniq = values.unique()
if len(uniq) == 1:
return uniq[0]
else:
return str(uniq[0]) + ", etc."
metadf = pd.pivot_table(df, index=series, aggfunc=aggfunc, dropna=False, sort=sort)
del metadf["value"]
if confidence_level is None:
df = pd.pivot_table(df, index=series, columns=groups, values='value', dropna=False, sort=sort)
return (df, None, metadf)
else:
def conf_intv(values):
return confidence_interval(confidence_level, values)
pivoted = pd.pivot_table(df, index=series, columns=groups, values="value", aggfunc=["mean", conf_intv], dropna=False, sort=sort)
valuedf = pivoted["mean"]
errorsdf = pivoted["conf_intv"]
if errorsdf.isna().values.all():
errorsdf = None
return (valuedf, errorsdf, metadf)
[docs]
def pivot_for_scatterchart(df, xaxis_itervar, group_by, confidence_level=None):
"""
Turns a DataFrame containing scalar results (in the format returned
by `results.get_scalars()`) into a DataFrame which can then be
passed to `utils.plot_lines()`.
Parameters:
- `df` (pandas.DataFrame): The dataframe to pivot.
- `xaxis_itervar` (string): The name of the iteration variable whose
values are to be used as X coordinates.
- `group_by` (list): A list of column names, the values in which will
be used to group the scalars into lines.
- `confidence_level` (float, optional):
The confidence level to use when computing the sizes of the error bars.
Returns:
- A DataFrame containing the pivoted data, with these columns:
`name`, `x`, `y`, and optionally `error` - if `confidence_level` is given.
"""
df = to_numeric(df, group_by + [xaxis_itervar])
newdf = pd.DataFrame()
if confidence_level is None:
df = pd.pivot_table(df, values="value", columns=group_by, index=xaxis_itervar)
errors_df = None
else:
def conf_intv(values):
return confidence_interval(confidence_level, values)
pivoted = pd.pivot_table(df, values="value", columns=group_by, index=xaxis_itervar if xaxis_itervar else "name", aggfunc=["mean", conf_intv], dropna=False)
df = pivoted["mean"]
errors_df = pivoted["conf_intv"]
if errors_df.isna().values.all():
errors_df = None
try:
xs = pd.to_numeric(df.index.values)
ideplot.xlabel(xaxis_itervar)
except:
xs = np.zeros_like(df.index.values)
for c in df:
t = c if type(c) == tuple else (c,)
name = ", ".join([str(a) + "=" + str(b) for a, b in zip(df.columns.names, t) if a is not None])
to_append = pd.DataFrame.from_dict({"name": [name], "x": [np.array(xs)], "y": [np.array(df[c].values)]})
if errors_df is not None:
to_append["error"] = [np.array(errors_df[c].values)]
newdf = pd.concat([newdf, to_append])
return newdf
[docs]
def get_confidence_level(props):
"""
Returns the confidence level from the `confidence_level` property,
converted to a `float`. Also accepts "none" (returns `None` in this case),
and percentage values (e.g. "95%").
"""
if "confidence_level" not in props:
return None
s = props["confidence_level"]
if s == "none":
return None
return float(s[:-1])/100 if s.endswith("%") else float(s)
def _perform_vector_op(df, line: str):
"""
Performs one vector operation on the dataframe. Helper for `perform_vector_ops()`.
"""
# parse line
type, module, name, args, kwargs, arglist_str = _parse_vectorop_line(line)
if name is None: # empty line
return df
# look up operation
function = vectorops.lookup_operation(module, name)
if function is None:
errmsg = "Vector filter function '" + name + "' not found"
if module is not None:
errmsg += " in module '" + module + "'"
raise ValueError(errmsg)
# perform operation
if type == "apply":
df = _apply_vector_op(df, name + arglist_str, function, *args, **kwargs)
elif type == "compute":
df = _compute_vector_op(df, name + arglist_str, function, *args, **kwargs)
return df
def _parse_vectorop_line(line: str):
"""
Parses the given vector operation line, and returns its contents as a tuple.
Helper for `perform_vector_ops()`.
"""
line = line.strip()
m = re.match(r"""(?x) # allow whitespace and comments in regex
((\w+)\s*:)? # optional "apply:" or "compute:"
\s* # optional whitespace
(([\w.]+)\.)? # optional qualifier (module prefix) for function name
(\w+)? # function name (optional so we can allow blank and comment-only lines)
(.*) # the rest: arglist in parens, comment (both optional)
""", line) # note: this regex always matches, and always matches the whole string due to last group
_, type, _, module, name, rest = m.groups()
if not name:
if line and not line.startswith('#'):
raise SyntaxError("Syntax error")
return (None, None, None, None, None, None)
if not type:
type = "apply"
if not type in ['apply', 'compute']:
raise SyntaxError("Syntax error near '"+type+"': must be 'apply' or 'compute' (or omitted)")
rest = rest.strip()
if not rest or rest.startswith('#'):
rest = "()"
if not rest.startswith('('):
raise ValueError("Parenthesized argument list expected after operation name")
try:
def return_args(*args, **kwargs):
return (args,kwargs)
args, kwargs = eval(name + " " + rest, None, {name: return_args}) # let Python do the parsing, incl. comment discarding
except SyntaxError:
raise SyntaxError("Syntax error in argument list")
except Exception as e:
raise ValueError("Error in argument list: " + str(e))
arglist_str = re.sub(r"\)\s*#.*", ")", rest) if args or kwargs else ""
return type, module, name, args, kwargs, arglist_str
def _apply_vector_op(df, op_str, operation, *args, **kwargs):
"""
Process a vector operation with the `apply` prefix. Helper for `perform_vector_ops()`.
"""
if operation == vectorops.aggregate:
return vectorops.aggregate(df, *args, **kwargs)
elif operation == vectorops.merge:
return vectorops.merge(df)
else:
condition = kwargs.pop('condition', None)
def process(row):
row["vectime"].flags.writeable = False
row["vecvalue"].flags.writeable = False
row = operation(row, *args, **kwargs)
row["name"] = row["name"] + ":" + op_str
if not row["vectime"].flags.owndata:
row["vectime"] = row["vectime"].copy()
row["vectime"].flags.writeable = True
if not row["vecvalue"].flags.owndata:
row["vecvalue"] = row["vecvalue"].copy()
row["vecvalue"].flags.writeable = True
return row
# Using iteration instead of df.transform because that catches the ChartScriptErrors
# raised from the vector operation functions internally, which we don't want.
clone = pd.DataFrame()
for _, row in df.iterrows():
if condition is None or condition(row):
pr = process(row)
clone = pd.concat([clone, pr.to_frame().T], ignore_index=True)
return clone
def _compute_vector_op(df, op_str, operation, *args, **kwargs):
"""
Process a vector operation with the `compute` prefix. Helper for `perform_vector_ops()`.
"""
return pd.concat([df, _apply_vector_op(df, op_str, operation, *args, **kwargs)])
[docs]
def set_plot_title(title, suggested_chart_name=None):
"""
Sets the plot title. It also sets the suggested chart name (the name that
the IDE offers when adding a temporary chart to the Analysis file.)
"""
ideplot.title(title)
chart.set_suggested_chart_name(suggested_chart_name if suggested_chart_name is not None else title)
[docs]
def fill_missing_titles(df):
"""
Utility function to fill missing values in the `title` and `moduledisplaypath`
columns from the `name` and `module` columns. (Note that `title` and `moduledisplaypath`
normally come from result attributes of the same name.)
"""
if "title" in df and "name" in df:
df["title"] = df["title"].fillna(df["name"])
if "moduledisplaypath" in df and "module" in df:
df["moduledisplaypath"] = df["moduledisplaypath"].fillna(df["module"])
[docs]
def extract_label_columns(df, props):
"""
Utility function to make a reasonable guess as to which column of
the given DataFrame is most suitable to act as a chart title and
which ones can be used as legend labels.
Ideally a "title column" should be one in which all lines have the same
value, and can be reasonably used as a title. This is often the `title`
or `name` column.
Label columns should be a minimal set of columns whose corresponding
value tuples uniquely identify every line in the DataFrame. These will
primarily be iteration variables and run attributes.
Returns:
A pair of a string and a list; the first value is the name of the
"title" column, and the second one is a list of pairs, each
containing the index and the name of a "label" column.
Example: `('title', [(8, 'numHosts'), (7, 'iaMean')])`
"""
def get_prop(k):
return props[k] if k in props else None
fill_missing_titles(df)
prefer_titles = get_prop('legend_prefer_result_titles') == 'true'
prefer_displaypaths = get_prop('legend_prefer_module_display_paths') == 'true'
name_column = "title" if ((prefer_titles or "name" not in df) and "title" in df) else "name"
module_column = "moduledisplaypath" if ((prefer_displaypaths or "module" not in df) and "moduledisplaypath" in df) else "module"
title_cols = []
legend_cols = []
if name_column in df:
if df[name_column].nunique() == 1:
title_cols.append(name_column)
else:
legend_cols.append(name_column)
if module_column in df:
if df[module_column].nunique() == 1:
title_cols.append(module_column)
else:
legend_cols.append(module_column)
if len(df) == 1:
title_cols = [c for c in [name_column, module_column] if c in df]
legend_cols = title_cols
return title_cols, legend_cols
if not title_cols:
title_col_candidates = ["experiment", "measurement", "replication"]
# for title, choose the first column that has has identical values in all rows
for col in title_col_candidates:
if col in df and col not in title_cols and df[col].nunique() == 1 and df[col].values[0]:
title_cols.append(col)
break
if not title_cols and name_column in df:
title_cols = [name_column]
blacklist = set(["name", "title", "module", "moduledisplaypath", # these will be used anyway
"runID", "value", "attrvalue", "vectime", "vecvalue",
"binedges", "binvalues", "underflows", "overflows",
"count", "sumweights", "mean", "stddev", "min", "max",
"processid", "datetime", "datetimef", "runnumber", "seedset",
"iterationvars", "iterationvarsf", "iterationvarsd", "repetition",
"source", "recordingmode", "interpolationmode", "enumname", "enum", "unit"])
# if unsuccessful, try to pick from all columns, except a few that we don't like
legend_col_candidates = [col for col in list(df.columns.values) if col not in blacklist and col not in title_cols]
# last resort columns
if "datetime" in df:
legend_col_candidates.append("datetime")
if "runID" in df:
legend_col_candidates.append("runID")
# pick columns that improve the partitioning of rows, until each row can be identified with them
last_len = None
for col in legend_col_candidates:
try:
new_len = len(df.groupby(legend_cols + [col]))
if new_len == len(df) or not last_len or new_len > last_len:
legend_cols.append(col)
last_len = new_len
if new_len == len(df):
break
except Exception:
pass
# filter out columns that only have a single value in them
# (this can happen in the loop above, with the first considered column)
legend_cols = list(filter(lambda icol: df[icol].nunique() > 1, legend_cols))
# at this point, ideally this should hold: len(df.groupby(legend_cols)) == len(df)
return title_cols, legend_cols
[docs]
def make_chart_title(df, title_cols):
"""
Produces a reasonably good chart title text from a result DataFrame,
given a selected list of "title" columns.
"""
if df is None or df.empty:
return "None"
def fmt(col):
if col in ["name", "title"]:
prefix = ""
elif col in ["module", "moduledisplaypath"]:
prefix = " in "
else:
prefix = ", " + col + "="
val = str(df[col].values[0])
if df[col].nunique() > 1:
val += ", etc."
return prefix + val
title = "".join([fmt(col) for col in title_cols if col in df])
title = _removeprefix(_removeprefix(title, ", "), " in ")
return title
[docs]
def select_best_partitioning_column_pair(df, props=None): #TODO remove default for props in final release
"""
Choose two columns from the dataframe which best partitions the rows
of the dataframe, and returns their names as a pair. Returns (`None`, `None`)
if no such pair was found. This method is useful for creating e.g. a bar plot.
"""
if props is None:
logger.warning("select_best_partitioning_column_pair(): Missing props argument! Update chart script, or code will break on next release!")
props = {}
title_cols, label_cols = extract_label_columns(df, props)
if len(label_cols) == 0:
return None, None
if len(label_cols) == 1:
if len(title_cols) == 0:
return None, None
elif label_cols[0] == title_cols[0]:
return None, None
else:
return title_cols[0], label_cols[0]
if len(label_cols) >= 2:
return label_cols[1], label_cols[0]
[docs]
def select_groups_series(df, props):
"""
Extracts the column names to be used for groups and series from the `df` DataFrame,
for pivoting. The columns whose names are to be used as group names are given in
the "groups" property in `props`, as a comma-separated list.
The names for the series are selected similarly, based on the "series" property.
There should be no overlap between these two lists.
If both "groups" and "series" are given (non-empty), they are simply returned
as lists after some sanity checks.
If both of them are empty, a reasonable guess is made for which columns should
be used, and (["module"], ["name"]) is used as a fallback.
The data in `df` should be in the format as returned by `result.get_scalars()`,
and the result can be used directly by `utils.pivot_for_barchart()`.
Returns:
- (group_names, series_names): A pair of lists of strings containing the
selected names for the groups and the series, respectively.
"""
groups = split(props["groups"])
series = split(props["series"])
if not groups and not series:
logger.info("The Groups and Series options were not set in the dialog, inferring them from the data.")
g, s = ("module", "name") if len(df) == 1 else select_best_partitioning_column_pair(df, props)
groups, series = [g], [s]
if not groups or not groups[0] or not series or not series[0]:
raise chart.ChartScriptError("Please set both the Groups and Series properties in the dialog - or neither, for automatic setup.")
common = list(set(groups) & set(series))
if common:
raise chart.ChartScriptError("Overlap between Group and Series columns: " + ", ".join(common))
assert_columns_exist(df, groups + series, "No such iteration variable or run attribute")
return groups, series
[docs]
def select_xaxis_and_groupby(df, props):
"""
Extracts an iteration variable name and the column names to be used for grouping from
the `df` DataFrame, for pivoting. The columns whose names are to be used as group
names are given in the "group_by" property in `props`, as a comma-separated list.
The name of the iteration variable is selected similarly, from the "xaxis_itervar" property.
The "group_by" list should not contain the given "xaxis_itervar" name.
If both "xaxis_itervar" and "group_by" are given (non-empty), they are simply returned
after some sanity checks, with "group_by" split into a list.
If both of them are empty, a reasonable guess is made for which columns should be used.
The data in `df` should be in the format as returned by `result.get_scalars()`,
and the result can be used directly by `utils.pivot_for_scatterchart()`.
Returns:
- (xaxis_itervar, group_by): An iteration variable name, and a list of strings
containing the selected column names to be used as groups.
"""
xaxis_itervar = props["xaxis_itervar"]
group_by = split(props["group_by"])
if not xaxis_itervar and not group_by:
logger.info("The 'X Axis' and 'Group By' options were not set in the dialog, inferring them from the data.")
xaxis_itervar, group_by = select_best_partitioning_column_pair(df, props)
group_by = [group_by] if group_by else []
logger.info(f"X Axis: {xaxis_itervar}, Group By: " + ",".join(group_by))
if xaxis_itervar:
assert_columns_exist(df, [xaxis_itervar], "The iteration variable for the X axis could not be found")
else:
raise chart.ChartScriptError("Please select the iteration variable for the X axis!")
if xaxis_itervar in group_by:
raise chart.ChartScriptError("X axis column also in grouper columns: " + xaxis_itervar)
if group_by:
assert_columns_exist(df, group_by, "An iteration variable for grouping could not be found")
# report averaging
uninteresting = ["runID", "value", "datetime", "datetimef", "processid",
"iterationvars", "iterationvarsf", "iterationvarsd",
"measurement", "replication", "runnumber", "seedset",
"title", "source", "interpolationmode"]
for c in df:
ul = len(df[c].unique())
if ul > 1 and c != xaxis_itervar and c not in group_by and c not in uninteresting:
logger.info(f"Points are averaged from an overall {ul} unique {c} values.")
return xaxis_itervar, group_by
[docs]
def assert_columns_exist(df, cols, message="Expected column missing from DataFrame"):
"""
Ensures that the dataframe contains the given columns. If any of them are missing,
the function raises an error with the given message.
Parameters:
- `df` (DataFrame): The DataFrame to operate on
- `cols` (list of strings): The list of column names to check.
"""
for c in cols:
if c not in df:
raise chart.ChartScriptError(message + ": " + c)
[docs]
def to_numeric(df, columns=list()):
"""
Convenience function. Runs `pandas.to_numeric` on the given
(or all) columns of `df`. If any of the given columns doesn't
exist, throws an error.
Parameters:
- `df` (DataFrame): The DataFrame to operate on
- `columns` (string or list of strings): The column name or list of
column names to convert. If not given, all columns will be converted.
"""
if not columns:
columns = list(df.columns.values)
df = df.copy()
if not isinstance(columns, list):
columns = [columns]
if columns:
assert_columns_exist(df, columns)
else:
columns = df.columns.values
for c in columns:
# Attempt to convert the entire column to numeric, but store the result separately
numeric_data = pd.to_numeric(df[c], errors='coerce')
# Update only the successfully converted (non-NaN) values back into the DataFrame
df.loc[numeric_data.notna(), c] = numeric_data[numeric_data.notna()]
return df
[docs]
def parse_rcparams(rc_content):
"""
Accepts a multiline string that contains rc file content in Matplotlib's
RcParams syntax, and returns its contents as a dictionary. Parse errors
and duplicate keys are reported via exceptions.
"""
rc_temp = {}
for line_no, line in enumerate(rc_content.split("\n"), 1):
strippedline = line.split('#', 1)[0].strip()
if not strippedline:
continue
tup = strippedline.split(':', 1)
if len(tup) != 2:
raise RuntimeError("Illegal rc line #" + str(line_no) + ": " + line)
key, val = tup
key = key.strip()
val = val.strip()
if key in rc_temp:
raise RuntimeError("Duplicate key " + key + " on line " + str(line_no))
rc_temp[key] = val
return rc_temp
def _filter_by_key_prefix(props, prefix):
return {k[len(prefix):] : v for (k, v) in props.items() if k.startswith(prefix) and v}
def _parse_optional_bool(value):
if value is None:
return None
if value.lower() not in ["true", "false"]:
raise ValueError("Invalid boolean property value: " + value)
return value.lower() == "true"
[docs]
def make_fancy_xticklabels(ax):
"""
Only useful for Matplotlib plots. It causes the x tick labels to be rotated
by the minimum amount necessary so that they don't overlap. Note that the
necessary amount of rotation typically depends on the horizontal zoom level.
"""
from matplotlib.text import Text
# Declare and register callbacks
def on_xlims_change(ax):
display_xs = list()
fontheight = 24
for p in ax.get_xticks():
xdisplay, ydisplay = ax.transData.transform_point((p, 0))
display_xs.append(xdisplay)
dxa = np.array(display_xs)
dxd = dxa[1:] - dxa[:-1]
min_dx = dxd.min() + 12
tes = list()
for l in ax.get_xticklabels():
text = Text(text=l, figure=ax.figure)
extent = text.get_window_extent()
tes.append(extent.width)
fontheight = extent.height
max_tx = np.array(tes).max()
if min_dx > max_tx:
angle = 0
elif min_dx < fontheight * 2:
angle = 90
else:
angle = math.atan(fontheight / (min_dx - fontheight * 2)) / math.pi * 180
angle = max(0, min(90, angle))
plt.xticks(rotation=angle, horizontalalignment="right")
ax.callbacks.connect('xlim_changed', on_xlims_change)
# inspired by: https://stackoverflow.com/a/11562898/635587
# and https://stackoverflow.com/q/11551049#comment77920445_11562898
def _make_scroll_navigable(figure):
"""
Only useful for Matplotlib plots. It causes the plot to respond to mouse
wheel events in the following way, regardless of navigation mode:
- Without modifiers: vertical pan
- With Shift: horizontal pan
- With Ctrl: zoom
"""
def zoom_fun(event):
ax = event.inaxes
if ax is None:
return
SCALING_FACTOR = 1.5
PANNING_FACTOR = 0.1
cur_xlim = ax.get_xlim()
cur_ylim = ax.get_ylim()
cur_xrange = cur_xlim[1] - cur_xlim[0]
cur_yrange = cur_ylim[1] - cur_ylim[0]
xdata = event.xdata
ydata = event.ydata
direction = np.sign(event.step)
# We add this to enum strip axes, see the HACK in _plot_enum.
is_enum_strip = hasattr(ax, "_colorbar")
if event.key == "control": # zoom
scale = math.pow(SCALING_FACTOR, direction)
ax.set_xlim([xdata - (xdata-cur_xlim[0]) / scale,
xdata + (cur_xlim[1]-xdata) / scale])
if not is_enum_strip:
ax.set_ylim([ydata - (ydata-cur_ylim[0]) / scale,
ydata + (cur_ylim[1]-ydata) / scale])
else: # pan
if event.key == "shift" or is_enum_strip: # horizontal pan
delta = cur_xrange * -direction * PANNING_FACTOR
ax.set_xlim([cur_xlim[0] + delta, cur_xlim[1] + delta])
else: # vertical pan
delta = cur_yrange * direction * PANNING_FACTOR
ax.set_ylim([cur_ylim[0] + delta, cur_ylim[1] + delta])
figure.canvas.draw_idle()
figure.canvas.mpl_connect('scroll_event', zoom_fun)
def _interpolationmode_to_drawstyle(interpolationmode, hasenum):
"""
Converts an OMNeT++-style interpolation constant ('none', 'linear',
'sample-hold', etc.) to Matplotlib draw styles.
"""
interp = interpolationmode if interpolationmode else 'sample-hold' if hasenum else None
if not interp:
ds = "default"
elif interp == "none":
ds = "none"
elif interp == "linear":
ds = "default"
elif interp == "sample-hold":
ds = "steps-post"
elif interp == "backward-sample-hold":
ds = 'steps-pre'
else:
logger.warning(f"Unknown interpolationmode: ${interp}")
ds = None
return ds
def _make_line_args(props, t, df):
global _marker_cycle
style = dict()
def get_prop(k):
return props[k] if k in props else None
if get_prop("linestyle"):
style["linestyle"] = get_prop("linestyle") # note: must precede 'drawstyle' setting
ds = get_prop("drawstyle")
if not ds or ds == "auto":
interpolationmode = t.interpolationmode if "interpolationmode" in df else None
hasenum = "enum" in df
ds = _interpolationmode_to_drawstyle(interpolationmode, hasenum)
if ds == "none":
style["linestyle"] = " "
style["drawstyle"] = "default" # or any valid value, doesn't matter
else:
style["drawstyle"] = ds if ds != "linear" else "default"
if get_prop("linecolor"):
style["color"] = get_prop("linecolor")
else:
style["color"] = next(_color_cycle)
if get_prop("linewidth"):
style["linewidth"] = get_prop("linewidth")
if not get_prop("marker") or get_prop("marker") == "auto":
style["marker"] = next(_marker_cycle)
elif get_prop("marker") == 'none':
style["marker"] = ' '
else:
style["marker"] = get_prop("marker")[0] # take first character only, as Matplotlib uses 1-char codes; this allows us to include a description
if get_prop("markersize"):
style["markersize"] = get_prop("markersize")
# Special case: not letting both the lines and the markers to become
# invisible automatically. Only if the user specifically asks for it.
if style["marker"] == ' ' and style["linestyle"] == ' ':
orig_ds = get_prop("drawstyle")
if not orig_ds or orig_ds == "auto":
style["marker"] = '.'
return style
def _make_histline_args(props, t, df): # ??? is t and df needed at all?
style = dict()
def get_prop(k):
return props[k] if k in props else None
if get_prop("color"):
style["color"] = get_prop("color")
else:
style["color"] = next(_color_cycle)
if get_prop("linewidth"):
style["linewidth"] = get_prop("linewidth")
ds = get_prop("drawstyle")
if ds == "Solid":
style["histtype"] = "stepfilled"
elif ds == "Outline":
style["histtype"] = "step"
style["density"] = _parse_optional_bool(get_prop("normalize"))
style["cumulative"] = _parse_optional_bool(get_prop("cumulative"))
return style
def _make_bar_args(props, df): # ??? is df needed at all? should we also get the column name?
style = dict()
def get_prop(k):
return props[k] if k in props else None
if get_prop("color"):
style["color"] = get_prop("color")
else:
style["color"] = next(_color_cycle)
return style
def _to_label(x):
"""
Internal. Turns various types of objects into a string, so they can be used as labels.
"""
if x is None:
return ""
elif isinstance(x, str):
return x
elif isinstance(x, tuple):
return ", ".join(map(str, list(x)))
elif isinstance(x, list):
return ", ".join(map(str, x))
else:
return str(x)
[docs]
def split(s, sep=','):
"""
Split a string with the given separator (by default with comma), trim
the surrounding whitespace from the items, and return the result as a
list. Returns an empty list for an empty or all-whitespace input string.
(Note that in contrast, `s.split(',')` will return an empty array,
even for `s=''`.)
"""
parts = s.split(sep)
parts = [p.strip() for p in parts]
if parts == ['']:
parts = []
return parts
def _split_by_types(df, types):
result = list()
for t in types:
mask = df['type'] == t
result.append(df[mask])
df = df[~mask]
result.append(df)
return result
def _append_metadata_columns(df, meta, suffix):
meta = pd.pivot_table(meta, index="runID", columns="attrname", values="attrvalue", aggfunc="first")
if not meta.empty:
df = df.join(meta, on="runID", rsuffix=suffix)
return df
def _select_param_assignments(config_entries_df):
names = config_entries_df["attrname"]
is_typename = names.str.endswith(".typename")
is_param = ~is_typename & names.str.match(r"^.*\.[^.-]+$")
result = config_entries_df.loc[is_param]
return result
def _pivot_results(df, include_attrs, include_runattrs, include_itervars, include_param_assignments, include_config_entries):
itervars, runattrs, configs, attrs, df = _split_by_types(df, ["itervar", "runattr", "config", "attr"])
params = _select_param_assignments(configs)
if attrs is not None and not attrs.empty:
attrs = pd.pivot_table(attrs, columns="attrname", aggfunc='first', index=["runID", "module", "name"], values="attrvalue")
if attrs is not None and not include_attrs:
if "unit" in attrs:
attrs = attrs[["unit"]]
else:
attrs = attrs[[]]
if not attrs.empty:
# this column is no longer needed, and it collided with the commonly used "type" result attribute in `merge`
df.drop(["type"], axis=1, inplace=True, errors="ignore")
df = df.merge(attrs, left_on=["runID", "module", "name"], right_index=True, how='left', suffixes=(None, "_attr"))
if include_itervars:
df = _append_metadata_columns(df, itervars, "_itervar")
if include_runattrs:
df = _append_metadata_columns(df, runattrs, "_runattr")
if include_config_entries:
df = _append_metadata_columns(df, configs, "_config")
if include_param_assignments and not include_config_entries:
df = _append_metadata_columns(df, params, "_param")
df.drop(['type', 'attrname', 'attrvalue'], axis=1, inplace=True, errors="ignore")
df.reset_index(inplace=True, drop=True)
return df
def _pivot_metadata(df, metadf, include_runattrs, include_itervars, include_param_assignments, include_config_entries):
itervars, runattrs, configs, remainder = _split_by_types(metadf, ["itervar", "runattr", "config"])
params = _select_param_assignments(configs)
assert(remainder.empty)
if include_itervars:
df = _append_metadata_columns(df, itervars, "_itervar")
if include_runattrs:
df = _append_metadata_columns(df, runattrs, "_runattr")
if include_config_entries:
df = _append_metadata_columns(df, configs, "_config")
if include_param_assignments and not include_config_entries:
df = _append_metadata_columns(df, params, "_param")
df = df.drop(['type', 'attrname', 'attrvalue'], axis=1, errors="ignore")
df.reset_index(inplace=True, drop=True)
return df