+ if test[u"type"] not in (u"NDRPDR", ):
+ logging.warning(f"Invalid test type: {test[u'type']}")
+ continue
+ if u"-pdr" in plot_title.lower():
+ ttype = u"PDR"
+ elif u"-ndr" in plot_title.lower():
+ ttype = u"NDR"
+ else:
+ logging.warning(
+ f"Invalid test type: {test[u'type']}"
+ )
+ continue
+ y_tmp_vals[test[u"parent"]][0].append(
+ test[u"latency"][ttype][u"direction1"][u"min"])
+ y_tmp_vals[test[u"parent"]][1].append(
+ test[u"latency"][ttype][u"direction1"][u"avg"])
+ y_tmp_vals[test[u"parent"]][2].append(
+ test[u"latency"][ttype][u"direction1"][u"max"])
+ y_tmp_vals[test[u"parent"]][3].append(
+ test[u"latency"][ttype][u"direction2"][u"min"])
+ y_tmp_vals[test[u"parent"]][4].append(
+ test[u"latency"][ttype][u"direction2"][u"avg"])
+ y_tmp_vals[test[u"parent"]][5].append(
+ test[u"latency"][ttype][u"direction2"][u"max"])
+ except (KeyError, TypeError) as err:
+ logging.warning(repr(err))
+
+ x_vals = list()
+ y_vals = list()
+ y_mins = list()
+ y_maxs = list()
+ nr_of_samples = list()
+ for key, val in y_tmp_vals.items():
+ name = re.sub(REGEX_NIC, u"", key.replace(u'-ndrpdr', u'').
+ replace(u'2n1l-', u''))
+ x_vals.append(name) # dir 1
+ y_vals.append(mean(val[1]) if val[1] else None)
+ y_mins.append(mean(val[0]) if val[0] else None)
+ y_maxs.append(mean(val[2]) if val[2] else None)
+ nr_of_samples.append(len(val[1]) if val[1] else 0)
+ x_vals.append(name) # dir 2
+ y_vals.append(mean(val[4]) if val[4] else None)
+ y_mins.append(mean(val[3]) if val[3] else None)
+ y_maxs.append(mean(val[5]) if val[5] else None)
+ nr_of_samples.append(len(val[3]) if val[3] else 0)
+
+ traces = list()
+ annotations = list()
+
+ for idx, _ in enumerate(x_vals):
+ if not bool(int(idx % 2)):
+ direction = u"West-East"
+ else:
+ direction = u"East-West"
+ hovertext = (
+ f"No. of Runs: {nr_of_samples[idx]}<br>"
+ f"Test: {x_vals[idx]}<br>"
+ f"Direction: {direction}<br>"
+ )
+ if isinstance(y_maxs[idx], float):
+ hovertext += f"Max: {y_maxs[idx]:.2f}uSec<br>"
+ if isinstance(y_vals[idx], float):
+ hovertext += f"Mean: {y_vals[idx]:.2f}uSec<br>"
+ if isinstance(y_mins[idx], float):
+ hovertext += f"Min: {y_mins[idx]:.2f}uSec"
+
+ if isinstance(y_maxs[idx], float) and isinstance(y_vals[idx], float):
+ array = [y_maxs[idx] - y_vals[idx], ]
+ else:
+ array = [None, ]
+ if isinstance(y_mins[idx], float) and isinstance(y_vals[idx], float):
+ arrayminus = [y_vals[idx] - y_mins[idx], ]
+ else:
+ arrayminus = [None, ]
+ traces.append(plgo.Scatter(
+ x=[idx, ],
+ y=[y_vals[idx], ],
+ name=x_vals[idx],
+ legendgroup=x_vals[idx],
+ showlegend=bool(int(idx % 2)),
+ mode=u"markers",
+ error_y=dict(
+ type=u"data",
+ symmetric=False,
+ array=array,
+ arrayminus=arrayminus,
+ color=COLORS[int(idx / 2)]
+ ),
+ marker=dict(
+ size=10,
+ color=COLORS[int(idx / 2)],
+ ),
+ text=hovertext,
+ hoverinfo=u"text",
+ ))
+ annotations.append(dict(
+ x=idx,
+ y=0,
+ xref=u"x",
+ yref=u"y",
+ xanchor=u"center",
+ yanchor=u"top",
+ text=u"E-W" if bool(int(idx % 2)) else u"W-E",
+ font=dict(
+ size=16,
+ ),
+ align=u"center",
+ showarrow=False
+ ))
+
+ try:
+ # Create plot
+ file_type = plot.get(u"output-file-type", u".html")
+ logging.info(f" Writing file {plot[u'output-file']}{file_type}.")
+ layout = deepcopy(plot[u"layout"])
+ if layout.get(u"title", None):
+ layout[u"title"] = f"<b>Latency:</b> {layout[u'title']}"
+ layout[u"annotations"] = annotations
+ plpl = plgo.Figure(data=traces, layout=layout)
+
+ # Export Plot
+ ploff.plot(
+ plpl,
+ show_link=False, auto_open=False,
+ filename=f"{plot[u'output-file']}{file_type}"
+ )
+ except PlotlyError as err:
+ logging.error(
+ f" Finished with error: {repr(err)}".replace(u"\n", u" ")
+ )
+ return
+
+
+def plot_tsa_name(plot, input_data):
+ """Generate the plot(s) with algorithm:
+ plot_tsa_name
+ specified in the specification file.
+
+ :param plot: Plot to generate.
+ :param input_data: Data to process.
+ :type plot: pandas.Series
+ :type input_data: InputData
+ """
+
+ # Transform the data
+ plot_title = plot.get(u"title", u"")
+ logging.info(
+ f" Creating data set for the {plot.get(u'type', u'')} {plot_title}."
+ )
+ data = input_data.filter_tests_by_name(
+ plot, params=[u"throughput", u"parent", u"tags", u"type"])
+ if data is None:
+ logging.error(u"No data.")
+ return
+
+ y_vals = OrderedDict()
+ for job in data:
+ for build in job:
+ for test in build:
+ if y_vals.get(test[u"parent"], None) is None:
+ y_vals[test[u"parent"]] = {
+ u"1": list(),
+ u"2": list(),
+ u"4": list()
+ }
+ try:
+ if test[u"type"] not in (u"NDRPDR",):
+ continue
+
+ if u"-pdr" in plot_title.lower():
+ ttype = u"PDR"
+ elif u"-ndr" in plot_title.lower():
+ ttype = u"NDR"
+ else:
+ continue
+
+ if u"1C" in test[u"tags"]:
+ y_vals[test[u"parent"]][u"1"]. \
+ append(test[u"throughput"][ttype][u"LOWER"])
+ elif u"2C" in test[u"tags"]:
+ y_vals[test[u"parent"]][u"2"]. \
+ append(test[u"throughput"][ttype][u"LOWER"])
+ elif u"4C" in test[u"tags"]:
+ y_vals[test[u"parent"]][u"4"]. \
+ append(test[u"throughput"][ttype][u"LOWER"])