← Files ClaraARCHIVED FILE
modules/set-overlap-analysis/vendor/modules/charting/plotting_utilities.py
44.1 KB · Oct 6, 2026 · 06:02 UTC
import copy
import logging
import math
import plotly.graph_objects as go
import polars as pl
from plotly.subplots import make_subplots
from modules.charting.adjust_position import get_y1_y0_values
from modules.charting.chart_helpers import check_if_values_too_close
from modules.charting.chart_primitives import (
divide_by_value_prefix,
get_max_and_min_value,
set_other_color_to_grey,
)
from modules.charting.draw_charts_utils import get_polars_value_at_index
from modules.charting.draw_waterfall import set_semantic_bar_color
from modules.data.common_data_utils import pivot_lazy, transform_lazy_df
from modules.utilities.config import (
get_config_params,
get_naming_params,
)
from modules.utilities.error_messages import add_app_message_to_paramdict
from modules.utilities.helpers import (
drop_columns,
duplicate_dataframe,
flatten_cols_polars,
get_dataset_specific_parameter,
unique,
)
from modules.utilities.utils import (
concat_aligned_lazy,
ensure_lazyframe,
get_schema_and_column_names,
)
def check_if_two_periods_in_distribution_chart(periodsArray):
if len(periodsArray) > 1:
period0, period1 = periodsArray[0], periodsArray[1]
else:
period0, period1 = periodsArray[0], ""
return period0, period1
def check_if_negative_bubble_size_values(
df: pl.DataFrame | pl.LazyFrame, chartDict: dict, paramDict: dict
) -> tuple[pl.DataFrame | pl.LazyFrame, dict]:
"""Exclude nonpositive bubble areas while retaining a plotting warning.
Supports both eager and lazy Polars data structures without collecting.
"""
namingParams = get_naming_params()
warningMessageType = namingParams["warningMessageType"]
plotChartsTabKey = namingParams["plotChartsTab"]
bubbleSizeKey = namingParams["bubbleSize"]
bubbleSizeDimension = chartDict[bubbleSizeKey]
lf = df.lazy() if isinstance(df, pl.DataFrame) else df
has_nonpositive = lf.select((pl.col(bubbleSizeDimension) <= 0).any()).collect(
engine="streaming"
)[0, 0]
if has_nonpositive:
message = (
f"{bubbleSizeDimension} bubble size column contains zero or negative values."
" Correspondent rows have been excluded for plotting."
)
paramDict = add_app_message_to_paramdict(
message,
warningMessageType,
plotChartsTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=0,
)
lf = lf.filter(pl.col(bubbleSizeDimension) > 0)
result = lf.collect() if isinstance(df, pl.DataFrame) else lf
return result, paramDict
def calculate_percentage_data_column_metric(
df: pl.DataFrame,
numeratorMetric: str,
denominatorMetric: str,
percentMetric: str,
numberOfMetrics: int,
sumColsArray: list[str],
chartDict: dict,
) -> tuple[pl.DataFrame, dict, list[str], int, list[str]]:
"""Create a percentage metric using Polars operations."""
namingParams = get_naming_params()
averageTotalValue = namingParams["averageTotalValue"]
dataColMetricName = namingParams["dataColMetricName"]
totalAverageValue = df[numeratorMetric].sum() / df[denominatorMetric].sum() * 100
df = df.with_columns(
(pl.col(numeratorMetric) / pl.col(denominatorMetric) * 100)
.fill_null(0)
.alias(percentMetric)
)
chartDict[averageTotalValue][percentMetric] = totalAverageValue
chartDict[dataColMetricName][percentMetric] = percentMetric
sumColsArray.append(percentMetric)
numberOfMetrics += 1
return df, chartDict, sumColsArray, numberOfMetrics, sumColsArray
def calculate_price_data_column_metric(
df: pl.DataFrame,
numeratorMetric: str,
denominatorMetric: str,
priceMetric: str,
numberOfMetrics: int,
sumColsArray: list[str],
chartDict: dict,
) -> tuple[pl.DataFrame, dict, list[str], int, list[str]]:
"""Create an average price metric using Polars."""
namingParams = get_naming_params()
averageTotalValue = namingParams["averageTotalValue"]
dataColMetricName = namingParams["dataColMetricName"]
totalAverageValue = df[numeratorMetric].sum() / df[denominatorMetric].sum()
df = df.with_columns(
(pl.col(numeratorMetric) / pl.col(denominatorMetric))
.fill_null(0)
.alias(priceMetric)
)
chartDict[averageTotalValue][priceMetric] = totalAverageValue
chartDict[dataColMetricName][priceMetric] = priceMetric
sumColsArray.append(priceMetric)
numberOfMetrics += 1
return df, chartDict, sumColsArray, numberOfMetrics, sumColsArray
def calculate_average_data_column_metric(
df: pl.DataFrame,
metric: str,
averageMetric: str,
countName: str,
numberOfMetrics: int,
sumColsArray: list[str],
chartDict: dict,
) -> tuple[pl.DataFrame, dict, list[str], int, list[str]]:
"""Create an average metric by dividing ``metric`` by ``countName``."""
namingParams = get_naming_params()
averageTotalValue = namingParams["averageTotalValue"]
dataColMetricName = namingParams["dataColMetricName"]
totalAverageValue = df[metric].sum() / df[countName].sum()
df = df.with_columns((pl.col(metric) / pl.col(countName)).alias(averageMetric))
chartDict[averageTotalValue][averageMetric] = totalAverageValue
chartDict[dataColMetricName][averageMetric] = averageMetric
sumColsArray.append(averageMetric)
numberOfMetrics += 1
return df, chartDict, sumColsArray, numberOfMetrics, sumColsArray
def purge_other_runs_from_chartdict(chartDictCopy, run):
namingParams = get_naming_params()
configParams = get_config_params()
drilldownParamsDictName = namingParams["drilldownParamsDictName"]
drilldownReportRunName = namingParams["drilldownReportRunName"]
chartDict = copy.deepcopy(chartDictCopy)
invertedEmojiNumberDict = configParams[namingParams["invertedEmojiNumberDict"]]
if drilldownParamsDictName in chartDict:
emojiNumber = run.replace(drilldownReportRunName + " ", "")
emojiNumber = emojiNumber.strip()
if emojiNumber in invertedEmojiNumberDict:
chosenRow = invertedEmojiNumberDict[emojiNumber]
if chosenRow in chartDict[drilldownParamsDictName]:
chartDict[drilldownParamsDictName] = {}
chartDict[drilldownParamsDictName][chosenRow] = chartDictCopy[
drilldownParamsDictName
][chosenRow]
elif str(chosenRow) in chartDict[drilldownParamsDictName]:
chartDict[drilldownParamsDictName] = {}
chartDict[drilldownParamsDictName][str(chosenRow)] = chartDictCopy[
drilldownParamsDictName
][str(chosenRow)]
return chartDict
def make_syn_plot_comment_dataset(inputFrameArray, chartDict):
"""
Convert a list of Polars LazyFrames into a final Polars LazyFrame
consistent with the logic of the original Pandas-based code.
"""
namingParams = get_naming_params()
itemName = namingParams["itemName"]
dimensionName = namingParams["dimensionName"]
plotValuesAsChoice = namingParams["plotValuesAsChoice"]
absolute = namingParams["absolute"]
percentName = namingParams["percentName"]
valueName = namingParams["valueName"]
outputFrameArray = []
for lazy_df in inputFrameArray:
# Transform each lazy DF individually
transformed_df = transform_lazy_df(
lazy_df,
chartDict,
dimensionName,
itemName,
plotValuesAsChoice,
absolute,
percentName,
valueName,
)
outputFrameArray.append(transformed_df)
# Concatenate all transformed lazy DFs vertically. This remains lazy.
final_df = concat_aligned_lazy(outputFrameArray)
# Ensure dimensionName is string-typed in the final result
final_df = final_df.with_columns(pl.col(dimensionName).cast(pl.Utf8))
return final_df
def aggregate_syn_plot_data(
chartDict,
metric,
frameArray,
synColumnArray,
aggregateOtherItemsName,
synColorArray,
mostRecentPeriod,
paramDict,
):
namingParams = get_naming_params()
notMetConditionValue = namingParams["notMetConditionValue"]
metConditionValue = namingParams["metConditionValue"]
synthesisPlot = namingParams["synthesisPlot"]
chartDict[synthesisPlot] = metConditionValue
synColumnArray = list(dict.fromkeys(synColumnArray))
df = concat_aligned_lazy(frameArray)
df = df.fill_nan(0)
colorArray = set_other_color_to_grey(
synColumnArray, aggregateOtherItemsName, synColorArray, chartDict, 0
)
return df, colorArray, chartDict, synColumnArray
def make_dic_to_color_first_bar(df, paramDict, chartDict, colorDict, run, count, array):
"""
if small multiples, must make array with dictionaries
to colors first bar based on if planned or previous data
"""
namingParams = get_naming_params()
configParams = get_config_params()
planStemArray = configParams[namingParams["planStemArray"]]
showInitialAndFinalValues = namingParams["showInitialAndFinalValues"]
varianceAmountName = namingParams["varianceAmountName"]
workColumn = namingParams["workColumn"]
drilldownReportRunName = namingParams["drilldownReportRunName"]
totalVarianceAggregation = namingParams["totalVarianceAggregation"]
marginVarianceAggregation = namingParams["marginVarianceAggregation"]
varianceAggregation = namingParams["varianceAggregation"]
isYearBeforePy = namingParams["isYearBeforePy"]
newArray = []
isExpectedData = False
firstLabel = df[workColumn][0].lower()
for element in planStemArray:
if element in firstLabel:
isExpectedData = True
initialAndFinalValuesCanBeShown = True
if varianceAggregation in chartDict:
if (
chartDict[varianceAggregation]
not in [totalVarianceAggregation, marginVarianceAggregation]
and drilldownReportRunName in run
):
initialAndFinalValuesCanBeShown = False
if (
showInitialAndFinalValues in chartDict
and chartDict[showInitialAndFinalValues]
and initialAndFinalValuesCanBeShown
):
firstBarColor, lineWidth, lineColor = set_semantic_bar_color(
isExpectedData, colorDict, paramDict
)
xrefValue, yrefValue, x1Value = (
"x" + str(count),
"y" + str(count),
df[varianceAmountName][0],
)
newArray = [
{
"type": "rect",
"fillcolor": firstBarColor,
"opacity": 1,
"line_width": lineWidth,
"line_color": lineColor,
"xref": xrefValue,
"yref": yrefValue,
"y0": -0.4,
"y1": 0.4,
"x0": 0,
"x1": x1Value,
}
]
array = array + newArray
return array
def reverse_waterfall_y_range(fig):
"""
get the waterfall items in the right order on the y axis
"""
fig.update_yaxes(autorange="reversed")
return fig
def make_dic_to_add_line(
df,
paramDict,
chartDict,
colorDict,
run,
count,
shapeArray,
x0Value,
x1Value,
numberOfCharts,
isArrow,
isPeriodZero,
countRows,
):
"""
in case of small multiples, we build the dictionary to add the line shapes
"""
namingParams = get_naming_params()
varianceAmountName = namingParams["varianceAmountName"]
showInitialAndFinalValues = namingParams["showInitialAndFinalValues"]
drilldownReportRunName = namingParams["drilldownReportRunName"]
totalVarianceAggregation = namingParams["totalVarianceAggregation"]
marginVarianceAggregation = namingParams["marginVarianceAggregation"]
varianceAggregation = namingParams["varianceAggregation"]
selectedPeriods = namingParams["selectedPeriods"]
workColumn = namingParams["workColumn"]
initialAndFinalValuesCanBeShown = True
if varianceAggregation in chartDict:
if (
chartDict[varianceAggregation]
not in [totalVarianceAggregation, marginVarianceAggregation]
and drilldownReportRunName in run
):
initialAndFinalValuesCanBeShown = False
newShapeArray = []
if (
showInitialAndFinalValues in chartDict
and chartDict[showInitialAndFinalValues]
and initialAndFinalValuesCanBeShown
):
firstBarColor, lineWidth, lineColor = (
colorDict["whiteColor"],
1,
colorDict["lightGreyColor"],
)
xrefValue, yrefValue = "x" + str(count), "y" + str(count)
df_lazy = ensure_lazyframe(df)
periodOneValue = get_polars_value_at_index(
df_lazy.filter(pl.col(workColumn) == chartDict[selectedPeriods][1]),
varianceAmountName,
0,
)
periodZeroValue = get_polars_value_at_index(df_lazy, varianceAmountName, 0)
if isArrow:
lineWidth = 2
if periodOneValue >= periodZeroValue:
lineColor = colorDict["greenColor"]
else:
lineColor = colorDict["redColor"]
y0Value, y1Value, yshift, lineColor = get_y1_y0_values(
numberOfCharts,
False,
isArrow,
count,
isPeriodZero,
lineColor,
chartDict,
countRows,
)
layerPlacement = "below"
newShapeArray = [
{
"type": "line",
"opacity": 1,
"line_width": lineWidth,
"line_color": lineColor,
"yref": "paper",
"xref": xrefValue,
"y0": y0Value,
"y1": y1Value,
"x0": x0Value,
"x1": x1Value,
"layer": layerPlacement,
},
]
shapeArray = shapeArray + newShapeArray
return shapeArray
def make_dic_to_add_annotation(
df,
paramDict,
chartDict,
colorDict,
run,
count,
shapeArray,
numberOfCharts,
isText,
isArrow,
countRows,
):
"""
in case of small multiples, we build the dictionary to add the annotations (arrow and text)
"""
namingParams = get_naming_params()
varianceAmountName = namingParams["varianceAmountName"]
showInitialAndFinalValues = namingParams["showInitialAndFinalValues"]
drilldownReportRunName = namingParams["drilldownReportRunName"]
totalVarianceAggregation = namingParams["totalVarianceAggregation"]
marginVarianceAggregation = namingParams["marginVariance"]
deltaName = namingParams["deltaName"]
varianceAggregation = namingParams["varianceAggregation"]
selectedPeriods = namingParams["selectedPeriods"]
workColumn = namingParams["workColumn"]
initialAndFinalValuesCanBeShown = True
if varianceAggregation in chartDict:
if (
chartDict[varianceAggregation]
not in [totalVarianceAggregation, marginVarianceAggregation]
and drilldownReportRunName in run
):
initialAndFinalValuesCanBeShown = False
newShapeArray = []
if (
showInitialAndFinalValues in chartDict
and chartDict[showInitialAndFinalValues]
and initialAndFinalValuesCanBeShown
):
df_lazy = ensure_lazyframe(df)
periodOneValue = get_polars_value_at_index(
df_lazy.filter(pl.col(workColumn) == chartDict[selectedPeriods][1]),
varianceAmountName,
0,
)
periodZeroValue = get_polars_value_at_index(df_lazy, varianceAmountName, 0)
xrefValue, yrefValue = "x" + str(count), "y" + str(count)
if periodOneValue >= periodZeroValue:
arrowColor = colorDict["greenColor"]
else:
arrowColor = colorDict["redColor"]
y0Value, y1Value, yshift, lineColor = get_y1_y0_values(
numberOfCharts,
isText,
isArrow,
count,
True,
arrowColor,
chartDict,
countRows,
)
if periodZeroValue != 0:
difference = periodOneValue - periodZeroValue
difference = divide_by_value_prefix(difference, chartDict, False)
difference = deltaName + " " + str(difference)
percentChange = ((periodOneValue - periodZeroValue) / periodZeroValue) * 100
if math.isnan(percentChange):
percentChange = deltaName + " nan"
else:
percentChange = "<i>(" + str(int(round(percentChange, 0))) + "%)</i>"
changevalue = difference + " " + percentChange
else:
percentChange = deltaName + " nan"
changevalue = percentChange
if isArrow:
arrowHead, arrowSize = 5, 1
text = None
xshift = 0
else:
arrowHead, arrowSize = 5, 1
text = changevalue
xshift = 30
periodValue = max(periodZeroValue, periodZeroValue)
newShapeArray = [
{
"showarrow": isArrow,
"arrowcolor": arrowColor,
"text": text,
"xshift": xshift,
"yshift": yshift,
"arrowhead": arrowHead,
"arrowsize": arrowSize,
"ay": y0Value,
"y": y0Value,
"yref": "paper",
"ax": periodValue,
"x": periodValue,
"xref": xrefValue,
"axref": xrefValue,
},
]
shapeArray = shapeArray + newShapeArray
return shapeArray
def _get_max_and_min_value_pl(
df: pl.LazyFrame,
metric: str,
chartDict: dict,
periodsArray: list[str],
) -> pl.LazyFrame:
"""Return a LazyFrame with min/max columns and a color flag."""
namingParams = get_naming_params()
minValue = namingParams["minValue"]
maxValue = namingParams["maxValue"]
colorName = namingParams["colorName"]
chosenChartKey = namingParams["chosenChart"]
multitierColumnChart = namingParams["multitierColumnChart"]
horizontalWaterfallChart = namingParams["horizontalWaterfallChart"]
discountName = namingParams["discountName"]
discountInPercentName = namingParams["discountInPercentName"]
indirectCostsName = namingParams["indirectCostsName"]
cogsName = namingParams["cogsName"]
chosenChart = chartDict[chosenChartKey]
reverseColorMetricsArray = [
discountName,
discountInPercentName,
indirectCostsName,
cogsName,
]
columns, schema = get_schema_and_column_names(df)
for column in periodsArray:
if column not in columns:
df = df.with_columns(pl.lit(None).alias(column))
if chosenChart not in [multitierColumnChart, horizontalWaterfallChart]:
df = df.with_columns(
pl.min_horizontal([pl.col(p) for p in periodsArray]).alias(minValue),
pl.max_horizontal([pl.col(p) for p in periodsArray]).alias(maxValue),
)
if metric not in reverseColorMetricsArray:
color_expr = (pl.col(periodsArray[0]) > pl.col(periodsArray[1])).cast(pl.Int8)
else:
color_expr = (pl.col(periodsArray[0]) <= pl.col(periodsArray[1])).cast(pl.Int8)
df = df.with_columns(color_expr.alias(colorName))
return df
def _check_if_values_too_close_pl(
df: pl.DataFrame, chartDict: dict, metric: str, periodOrder: list[str]
) -> tuple[pl.DataFrame, str]:
"""Polars version of ``check_if_values_too_close``."""
namingParams = get_naming_params()
minValue = namingParams["minValue"]
maxValue = namingParams["maxValue"]
workColumn = namingParams["workColumn"]
hideLabel = namingParams["hideLabel"]
plotValuesAsChoice = namingParams["plotValuesAsChoice"]
absolute = namingParams["absolute"]
labelToHide = namingParams["labelToHide"]
pricePerUnitName = namingParams["pricePerUnitName"]
if chartDict[plotValuesAsChoice] == absolute and metric != pricePerUnitName:
chartFormat = ",.3s"
hideLabelLimit = 5
elif chartDict[plotValuesAsChoice] == absolute and metric == pricePerUnitName:
chartFormat = ",.2s"
hideLabelLimit = 1
elif metric == pricePerUnitName:
chartFormat = ",.1f"
hideLabelLimit = 1
else:
chartFormat = ",.0f"
hideLabelLimit = 1
total_max = df[maxValue].sum()
df = df.with_columns(
pl.lit(periodOrder[0]).alias(labelToHide),
((pl.col(maxValue) - pl.col(minValue)) / total_max * 100).alias(workColumn),
)
df = df.with_columns(
pl.when(pl.col(periodOrder[1]).is_null())
.then(periodOrder[1])
.otherwise(pl.col(labelToHide))
.alias(labelToHide)
)
df = df.with_columns(
pl.when((pl.col(periodOrder[0]).is_null()) | (pl.col(periodOrder[1]).is_null()))
.then(hideLabelLimit)
.otherwise(pl.col(workColumn))
.alias(workColumn)
)
df = df.with_columns(
pl.when(pl.col(workColumn) < hideLabelLimit)
.then(1)
.otherwise(0)
.alias(hideLabel)
)
df = df.drop(workColumn)
return df, chartFormat
def tag_if_increasing_or_decreasing(
dfCopy: pl.DataFrame,
metric: str,
chosenDimension: str,
paramDict: dict,
chartDict: dict,
) -> tuple[pl.DataFrame, str, dict]:
"""Prepare dumbbell chart data tagging increases/decreases."""
namingParams = get_naming_params()
periodName = namingParams["periodName"]
selectedPeriods = namingParams["selectedPeriods"]
separatorString = namingParams["separatorString"]
minValue = namingParams["minValue"]
maxValue = namingParams["maxValue"]
colorName = namingParams["colorName"]
labelName = namingParams["labelName"]
periodName = namingParams["periodName"]
hideLabel = namingParams["hideLabel"]
labelToHide = namingParams["labelToHide"]
periodOrder = chartDict[selectedPeriods]
df_lazy = duplicate_dataframe(dfCopy)
checkedPeriodOrder = []
if isinstance(dfCopy, pl.LazyFrame):
periodValues = (
dfCopy.select(pl.col(periodName).unique())
.collect(engine="streaming")
.get_column(periodName)
.to_list()
)
else:
periodValues = dfCopy[periodName].unique().to_list()
for period in periodOrder:
periodLower = period.lower()
periodUpper = period.upper()
if period in periodValues:
checkedPeriodOrder.append(period)
elif periodLower in periodValues:
checkedPeriodOrder.append(periodLower)
elif periodUpper in periodValues:
checkedPeriodOrder.append(periodUpper)
chartDict[selectedPeriods] = checkedPeriodOrder
df_lazy = df_lazy.group_by([chosenDimension, periodName]).agg(
pl.col(metric).sum().alias(metric)
)
df_lazy = pivot_lazy(
lf=df_lazy,
index_col=chosenDimension,
pivot_col=periodName,
value_col=metric,
agg_func="first",
)
df_lazy = flatten_cols_polars(df_lazy, separatorString)
columns, _ = get_schema_and_column_names(df_lazy)
rename_map = {
f"{metric}_{period}": period
for period in checkedPeriodOrder
if f"{metric}_{period}" in columns and period not in columns
}
if rename_map:
df_lazy = df_lazy.rename(rename_map)
df_lazy = _get_max_and_min_value_pl(df_lazy, metric, chartDict, checkedPeriodOrder)
pl_df = df_lazy.collect()
pl_df, chartFormat = _check_if_values_too_close_pl(
pl_df, chartDict, metric, checkedPeriodOrder
)
pl_df = pl_df.unpivot(
index=[
chosenDimension,
minValue,
maxValue,
colorName,
hideLabel,
"labelToHide",
],
variable_name=periodName,
value_name=metric,
)
pl_df = pl_df.with_columns(
pl.when(
(pl.col(hideLabel) == 1) & (pl.col(periodName) == pl.col("labelToHide"))
)
.then(None)
.otherwise(pl.col(metric))
.alias(labelName)
)
pl_df = drop_columns(pl_df, ["labelToHide", hideLabel])
return pl_df, chartFormat, chartDict
def make_df_counts_unique_values(
dfCopy: pl.DataFrame, countName: str, chartDict: dict
) -> pl.DataFrame:
"""Aggregate counts of unique values using Polars."""
namingParams = get_naming_params()
aggregateUniquesByDimension = namingParams["aggregateUniquesByDimension"]
countColumn = namingParams["countColumn"]
aggregateUniquesDimension = namingParams["aggregateUniquesDimension"]
dfCounts = duplicate_dataframe(dfCopy)
if chartDict[aggregateUniquesByDimension]:
dim_col = chartDict[aggregateUniquesDimension]
dfCounts = (
dfCounts.select([chartDict[countColumn], dim_col])
.with_columns(pl.lit(1).alias(countName))
.with_columns(pl.col(dim_col).str.to_titlecase())
.unique()
.group_by(dim_col)
.agg(pl.col(countName).sum())
)
else:
dim_col = chartDict[countColumn]
dfCounts = (
dfCounts.select([dim_col])
.with_columns(pl.lit(1).alias(countName))
.with_columns(pl.col(dim_col).str.to_titlecase())
.unique()
.group_by(dim_col)
.agg(pl.col(countName).sum())
)
return dfCounts
def make_df_for_pareto_classes(df, dfCounts):
namingParams = get_naming_params()
className = namingParams["className"]
group_byCols = [className]
indexColumn = className
df = df.join(dfCounts, on=className, how="left")
return df, group_byCols, indexColumn
def make_df_for_pareto_items(
df: pl.DataFrame | pl.LazyFrame,
dfCounts: pl.DataFrame | pl.LazyFrame,
countName: str,
chartDict: dict,
) -> tuple[pl.LazyFrame, list[str], str]:
"""Return ``df`` joined with counts for Pareto charts as a LazyFrame."""
namingParams = get_naming_params()
metricsToPlot = namingParams["metricsToPlot"]
countColumn = namingParams["countColumn"]
aggregateUniquesDimension = namingParams["aggregateUniquesDimension"]
metricsToPlot = chartDict[metricsToPlot]
group_byCols = [chartDict[aggregateUniquesDimension]]
indexColumn = chartDict[aggregateUniquesDimension]
lf_items = ensure_lazyframe(df)
lf_counts = ensure_lazyframe(dfCounts)
dfItems = lf_items.select([chartDict[aggregateUniquesDimension], metricsToPlot[0]])
lf_counts = lf_counts.join(
dfItems, on=chartDict[aggregateUniquesDimension], how="full"
)
otherRank = lf_counts.filter(
pl.col(countName).is_null() & pl.col(metricsToPlot[0]).is_not_null()
)
has_other = otherRank.select(pl.len()).collect().item() > 0
if has_other:
other_value = (
otherRank.select(pl.col(chartDict[aggregateUniquesDimension]).first())
.collect()
.item()
)
lf_counts = lf_counts.with_columns(
pl.when(pl.col(metricsToPlot[0]).is_null())
.then(pl.lit(other_value))
.otherwise(pl.col(chartDict[countColumn]))
.alias(chartDict[countColumn])
)
lf_counts = (
lf_counts.select([chartDict[aggregateUniquesDimension], countName])
.drop_nulls()
.group_by(chartDict[aggregateUniquesDimension])
.agg(pl.col(countName).sum())
)
lf_counts = drop_columns(lf_counts, [metricsToPlot[0]])
lf_result = lf_items.join(
lf_counts, on=chartDict[aggregateUniquesDimension], how="left"
)
return lf_result, group_byCols, indexColumn
def get_pareto_axis(fig, metricsToPlot, chartDict):
namingParams = get_naming_params()
fixedScaleChoice = namingParams["fixedParetoScaleChoice"]
scaleOne, scaleTwo, scaleThree, scaleFour, scaleFive, scaleSix = (
False,
False,
False,
False,
False,
False,
)
if (
len(metricsToPlot) > 1
and fixedScaleChoice in chartDict
and chartDict[fixedScaleChoice]
):
fullFig = fig.full_figure_for_development(warn=False)
scaleOne = fullFig.layout.xaxis.range[1]
if "xaxis2" in fullFig.layout:
scaleTwo = fullFig.layout.xaxis2.range[1]
if len(metricsToPlot) > 2 and "xaxis3" in fullFig.layout:
scaleThree = fullFig.layout.xaxis3.range[1]
if len(metricsToPlot) > 3 and "xaxis4" in fullFig.layout:
scaleFour = fullFig.layout.xaxis4.range[1]
if len(metricsToPlot) > 4 and "xaxis5" in fullFig.layout:
scaleFive = fullFig.layout.xaxis5.range[1]
if len(metricsToPlot) > 5 and "xaxis6" in fullFig.layout:
scaleSix = fullFig.layout.xaxis6.range[1]
scaleArray = [scaleOne, scaleTwo, scaleThree, scaleFour, scaleFive, scaleSix]
maxScale = max(scaleArray)
fig.update_layout(xaxis_range=[fullFig.layout.xaxis.range[0], maxScale])
if "xaxis2" in fullFig.layout:
fig.update_layout(xaxis2_range=[fullFig.layout.xaxis2.range[0], maxScale])
if len(metricsToPlot) > 2 and "xaxis3" in fullFig.layout:
fig.update_layout(xaxis3_range=[fullFig.layout.xaxis3.range[0], maxScale])
if len(metricsToPlot) > 3 and "xaxis4" in fullFig.layout:
fig.update_layout(xaxis4_range=[fullFig.layout.xaxis4.range[0], maxScale])
if len(metricsToPlot) > 4 and "xaxis5" in fullFig.layout:
fig.update_layout(xaxis5_range=[fullFig.layout.xaxis5.range[0], maxScale])
if len(metricsToPlot) > 5 and "xaxis6" in fullFig.layout:
fig.update_layout(xaxis6_range=[fullFig.layout.xaxis6.range[0], maxScale])
return fig
def join_metric_dataframes(
dfDict: dict[str, pl.LazyFrame | pl.DataFrame], metricsToPlot: list[str]
) -> pl.LazyFrame:
"""Left join metric frames using Polars lazily."""
base_key = metricsToPlot[0]
base_df = ensure_lazyframe(dfDict[base_key]).clone()
columns, _ = get_schema_and_column_names(base_df)
join_col = columns[0]
for key in metricsToPlot[1:]:
other_df = ensure_lazyframe(dfDict[key]).clone()
columns, _ = get_schema_and_column_names(other_df)
other_df = other_df.rename({columns[0]: join_col})
base_df = base_df.join(other_df, on=join_col, how="left")
return base_df
def calculate_metrics_for_data_column(df, chartDict, sumColsArray, countName):
"""
calculate metrics for Pareto stacked bar plot data column
"""
namingParams = get_naming_params()
countColumn = namingParams["countColumn"]
dataColMetricName = namingParams["dataColMetricName"]
averageTotalValue = namingParams["averageTotalValue"]
rankingMetric = namingParams["rankingMetric"]
showMetricsInDataColumn = namingParams["showMetricsInDataColumn"]
metricsToShowInDataColumn = namingParams["metricsToShowInDataColumn"]
monetaryLocalCurrencyName = namingParams["monetaryLocalCurrencyName"]
averageAmount = namingParams["averageAmount"]
unitsName = namingParams["unitsName"]
volumeName = namingParams["volumeName"]
averageUnits = namingParams["averageUnits"]
averageVolume = namingParams["averageVolume"]
discountName = namingParams["discountName"]
pricePerUnitName = namingParams["pricePerUnitName"]
pricePerVolumeName = namingParams["pricePerVolumeName"]
discountInPercentName = namingParams["discountInPercentName"]
netOfDiscountName = namingParams["netOfDiscountName"]
averageAmountAfterDiscount = namingParams["averageAmountAfterDiscount"]
marginName = namingParams["marginName"]
averageMargin = namingParams["averageMargin"]
nothingThereString = namingParams["nothingThereString"]
marginInPercentName = namingParams["marginInPercentName"]
marginInPercentOfNetSalesName = namingParams["marginInPercentOfNetSalesName"]
pricePerUnitNetDiscountName = namingParams["pricePerUnitNetDiscountName"]
pricePerVolumeNetDiscountName = namingParams["pricePerVolumeNetDiscountName"]
averageTotalValue = namingParams["averageTotalValue"]
dataColMetricName = namingParams["dataColMetricName"]
numberOfMetrics = 0
columns, schema = get_schema_and_column_names(df)
numberOfColsAtStart = len(columns)
chartDict = drop_choices_over_max(chartDict)
chartDict[averageTotalValue] = {}
chartDict[dataColMetricName] = {}
if showMetricsInDataColumn in chartDict and chartDict[showMetricsInDataColumn]:
if averageAmount in chartDict[metricsToShowInDataColumn]:
df, chartDict, sumColsArray, numberOfMetrics, sumColsArray = (
calculate_average_data_column_metric(
df,
monetaryLocalCurrencyName,
averageAmount,
countName,
numberOfMetrics,
sumColsArray,
chartDict,
)
)
if averageUnits in chartDict[metricsToShowInDataColumn]:
df, chartDict, sumColsArray, numberOfMetrics, sumColsArray = (
calculate_average_data_column_metric(
df,
unitsName,
averageUnits,
countName,
numberOfMetrics,
sumColsArray,
chartDict,
)
)
if averageVolume in chartDict[metricsToShowInDataColumn]:
df, chartDict, sumColsArray, numberOfMetrics, sumColsArray = (
calculate_average_data_column_metric(
df,
volumeName,
averageVolume,
countName,
numberOfMetrics,
sumColsArray,
chartDict,
)
)
if pricePerUnitName in chartDict[metricsToShowInDataColumn]:
df, chartDict, sumColsArray, numberOfMetrics, sumColsArray = (
calculate_price_data_column_metric(
df,
monetaryLocalCurrencyName,
unitsName,
pricePerUnitName,
numberOfMetrics,
sumColsArray,
chartDict,
)
)
if pricePerVolumeName in chartDict[metricsToShowInDataColumn]:
df, chartDict, sumColsArray, numberOfMetrics, sumColsArray = (
calculate_price_data_column_metric(
df,
monetaryLocalCurrencyName,
volumeName,
pricePerVolumeName,
numberOfMetrics,
sumColsArray,
chartDict,
)
)
if pricePerUnitNetDiscountName in chartDict[metricsToShowInDataColumn]:
df, chartDict, sumColsArray, numberOfMetrics, sumColsArray = (
calculate_price_data_column_metric(
df,
netOfDiscountName,
unitsName,
pricePerUnitNetDiscountName,
numberOfMetrics,
sumColsArray,
chartDict,
)
)
if pricePerVolumeNetDiscountName in chartDict[metricsToShowInDataColumn]:
df, chartDict, sumColsArray, numberOfMetrics, sumColsArray = (
calculate_price_data_column_metric(
df,
netOfDiscountName,
volumeName,
pricePerVolumeNetDiscountName,
numberOfMetrics,
sumColsArray,
chartDict,
)
)
if discountInPercentName in chartDict[metricsToShowInDataColumn]:
df, chartDict, sumColsArray, numberOfMetrics, sumColsArray = (
calculate_percentage_data_column_metric(
df,
discountName,
monetaryLocalCurrencyName,
discountInPercentName,
numberOfMetrics,
sumColsArray,
chartDict,
)
)
if averageAmountAfterDiscount in chartDict[metricsToShowInDataColumn]:
df, chartDict, sumColsArray, numberOfMetrics, sumColsArray = (
calculate_average_data_column_metric(
df,
netOfDiscountName,
averageAmountAfterDiscount,
countName,
numberOfMetrics,
sumColsArray,
chartDict,
)
)
if averageMargin in chartDict[metricsToShowInDataColumn]:
df, chartDict, sumColsArray, numberOfMetrics, sumColsArray = (
calculate_average_data_column_metric(
df,
marginName,
averageMargin,
countName,
numberOfMetrics,
sumColsArray,
chartDict,
)
)
if marginInPercentName in chartDict[metricsToShowInDataColumn]:
df, chartDict, sumColsArray, numberOfMetrics, sumColsArray = (
calculate_percentage_data_column_metric(
df,
marginName,
monetaryLocalCurrencyName,
marginInPercentName,
numberOfMetrics,
sumColsArray,
chartDict,
)
)
if marginInPercentOfNetSalesName in chartDict[metricsToShowInDataColumn]:
df, chartDict, sumColsArray, numberOfMetrics, sumColsArray = (
calculate_percentage_data_column_metric(
df,
netOfDiscountName,
monetaryLocalCurrencyName,
marginInPercentOfNetSalesName,
numberOfMetrics,
sumColsArray,
chartDict,
)
)
columns, schema = get_schema_and_column_names(df)
numberOfColsAtEnd = len(columns)
df = df.filter(pl.col(columns[0]) != nothingThereString)
return df, chartDict, sumColsArray
def calculate_actual_vs_previous_year_index_change(
df: pl.DataFrame, column: str | None, valueCols: list[str], paramDict: dict
) -> pl.DataFrame:
"""Calculate year-over-year index change using Polars."""
namingParams = get_naming_params()
acpyName = namingParams["acpyName"]
periodName = namingParams["periodName"]
acName = namingParams["acName"]
pyName = namingParams["pyName"]
yoyChangeName = namingParams["yoyChangeName"]
divideIndex = [periodName, acpyName]
if column:
divideIndex.append(column)
dfTy = df.filter(pl.col(acpyName) == acName).with_columns(
pl.lit(yoyChangeName).alias(acpyName)
)
dfYa = df.filter(pl.col(acpyName) == pyName).with_columns(
pl.lit(yoyChangeName).alias(acpyName)
)
joined = dfTy.join(dfYa, on=divideIndex, how="inner", suffix="_ya")
for col_name in valueCols:
joined = joined.with_columns(
pl.when(pl.col(f"{col_name}_ya") != 0)
.then((pl.col(col_name) / pl.col(f"{col_name}_ya") * 100).round(0))
.otherwise(0)
.alias(col_name)
)
result_cols = divideIndex + valueCols
return joined.select(result_cols)
def get_mins_and_maxes(dataArray, chartDict):
namingParams = get_naming_params()
startAxesFromZero = namingParams["startAxesFromZero"]
yAxisMetric = namingParams["yAxisMetric"]
xAxisMetric = namingParams["xAxisMetric"]
minXDimensionKey = namingParams["minXDimension"]
maxXDimensionKey = namingParams["maxXDimension"]
minYDimensionKey = namingParams["minYDimension"]
maxYDimensionKey = namingParams["maxYDimension"]
notMetConditionValue = namingParams["notMetConditionValue"]
xDimension = chartDict[xAxisMetric]
yDimension = chartDict[yAxisMetric]
if startAxesFromZero in chartDict and chartDict[startAxesFromZero]:
chartDict[minXDimensionKey], chartDict[maxXDimensionKey] = (
notMetConditionValue,
notMetConditionValue,
)
chartDict[minYDimensionKey], chartDict[maxYDimensionKey] = (
notMetConditionValue,
notMetConditionValue,
)
else:
chartDict[minXDimensionKey], chartDict[maxXDimensionKey] = (
99999999999999,
-99999999999999,
)
chartDict[minYDimensionKey], chartDict[maxYDimensionKey] = (
99999999999999,
-99999999999999,
)
for df in dataArray:
minXDimension = min(df[xDimension])
maxXDimension = max(df[xDimension])
minYDimension = min(df[yDimension])
maxYDimension = max(df[yDimension])
if minXDimension < chartDict[minXDimensionKey]:
chartDict[minXDimensionKey] = minXDimension
if maxXDimension > chartDict[maxXDimensionKey]:
chartDict[maxXDimensionKey] = maxXDimension
if minYDimension < chartDict[minYDimensionKey]:
chartDict[minYDimensionKey] = minYDimension
if maxYDimension > chartDict[maxYDimensionKey]:
chartDict[maxYDimensionKey] = maxYDimension
return chartDict
def drop_choices_over_max(chartDict):
namingParams = get_naming_params()
configParams = get_config_params()
numberOfMetricsInDataColumnKey = namingParams["numberOfMetricsInDataColumn"]
metricsToShowInDataColumn = namingParams["metricsToShowInDataColumn"]
maxNumberOfDataColMetrics = configParams[namingParams["maxNumberOfDataColMetrics"]]
numberOfMetricsInDataColumn = len(chartDict[metricsToShowInDataColumn])
if numberOfMetricsInDataColumn > maxNumberOfDataColMetrics:
chartDict[metricsToShowInDataColumn] = chartDict[metricsToShowInDataColumn][
0:maxNumberOfDataColMetrics
]
numberOfMetricsInDataColumn = len(chartDict[metricsToShowInDataColumn])
chartDict[numberOfMetricsInDataColumnKey] = numberOfMetricsInDataColumn
return chartDict
def make_integer_date_dict(uniqueDates):
dateToIntDict = {}
intToDateDict = {}
orderedList = list(range(1, len(uniqueDates) + 1))
count = 0
for date in uniqueDates:
intToDateDict[orderedList[count]] = date
dateToIntDict[date] = [orderedList[count]]
count = count + 1
return orderedList, intToDateDict, dateToIntDict
def delete_black_vertical_lines(fig):
fig.update_layout(
yaxis=dict(
showline=False, # Hides the vertical axis line but keeps the labels
showticklabels=True, # Ensures tick labels like "PL", "Price" remain
)
)
fig.update_yaxes(showline=False)
fig.update_yaxes(
zeroline=True, # Ensures the zero line is visible
zerolinecolor="black", # Changes the color of the zero line to black
zerolinewidth=2, # Optionally, you can adjust the thickness of the zero line
)
return fig
def set_axes_to_log(fig, chartDict):
namingParams = get_naming_params()
logXAxis = namingParams["logXAxis"]
logYAxis = namingParams["logYAxis"]
if logXAxis in chartDict and chartDict[logXAxis]:
fig.update_xaxes(type="log")
if logYAxis in chartDict and chartDict[logYAxis]:
fig.update_yaxes(type="log")
return fig
def set_number_of_cols_for_bubble_and_scatter_chart(smallMultipleUniqueItems):
numberOfCols = 2
if len(smallMultipleUniqueItems) <= 4:
numberOfCols = 2
elif len(smallMultipleUniqueItems) <= 6:
numberOfCols = 3
else:
numberOfCols = 4
return numberOfCols
def extract_values_from_dictionary(valueDict):
paramDict = valueDict["1"]
df = valueDict["2"]
dfDates = valueDict["3"]
dfPeriods = valueDict["4"]
dfAllPeriods = valueDict["5"]
dfPlan = valueDict["6"]
indexCols = valueDict["7"]
valueCols = valueDict["8"]
chartDict = valueDict["9"]
toDrop = valueDict["10"]
originalValueCols = valueDict["11"]
colDict = valueDict["12"]
tabDict = valueDict["13"]
automateDict = valueDict["14"]
planPlaybackDict = valueDict["15"]
return (
paramDict,
df,
dfDates,
dfPeriods,
dfAllPeriods,
dfPlan,
indexCols,
valueCols,
chartDict,
toDrop,
originalValueCols,
colDict,
tabDict,
automateDict,
planPlaybackDict,
)
SHA-256: 62a9e8e68fb02c56949b7e3b67822242fdbca5eb4ad5d05aa39f0b8fe51beb06