← Files VeraARCHIVED FILE
modules/variance-analysis/vendor/modules/data/common_data_utils.py
128 KB · Oct 2, 2026 · 00:29 UTC
import base64
import colorsys
import copy
import io
import json
# All data manipulation should rely on polars
import logging
import math
import random
import sys
from io import BytesIO
from typing import Any, Dict, List
import numpy as np
import polars as pl
from modules.layout.memoization import check_collect
from modules.utils.polars_excel_writer import write_polars_excel
if "modules.utilities.utils" in sys.modules:
utils_mod = sys.modules["modules.utilities.utils"]
else:
import modules.utilities.utils as utils_mod
from modules.utilities.config import (
get_config_params,
get_metric_array_params,
get_naming_params,
)
from modules.utilities.ui_notifier import ui as notifier
try:
from modules.utilities.helpers import (
add_running_total,
check_and_clean_columns,
check_and_group_by_cols,
check_if_periods_in_columns,
drop_columns,
duplicate_dataframe,
flatten_cols_polars,
get_period_length,
get_periods_array,
take_filtered_value_out_of_option_list,
unique,
)
except ImportError as e: # pragma: no cover - allow partial stubs in tests
import types
logging.getLogger(__name__).warning("common_data_utils helpers import error: %s", e)
_helpers = sys.modules.get(
"modules.utilities.helpers", types.ModuleType("modules.utilities.helpers")
)
add_running_total = getattr(_helpers, "add_running_total", lambda df: df)
check_and_clean_columns = getattr(
_helpers,
"check_and_clean_columns",
lambda df, group_cols, value_cols: (group_cols, value_cols),
)
check_and_group_by_cols = getattr(
_helpers,
"check_and_group_by_cols",
lambda df, group_cols, value_cols: (df, group_cols, value_cols),
)
check_if_periods_in_columns = getattr(
_helpers, "check_if_periods_in_columns", lambda *a, **k: False
)
drop_columns = getattr(_helpers, "drop_columns", lambda df, cols: df)
duplicate_dataframe = getattr(_helpers, "duplicate_dataframe", lambda df: df)
flatten_cols_polars = getattr(
_helpers, "flatten_cols_polars", lambda df, delim="": df
)
get_period_length = getattr(_helpers, "get_period_length", lambda *a, **k: 0)
get_periods_array = getattr(_helpers, "get_periods_array", lambda *a, **k: [])
take_filtered_value_out_of_option_list = getattr(
_helpers,
"take_filtered_value_out_of_option_list",
lambda arr, val: arr,
)
unique = getattr(_helpers, "unique", lambda arr: list(dict.fromkeys(arr)))
ensure_lazyframe = utils_mod.ensure_lazyframe
get_schema_and_column_names = utils_mod.get_schema_and_column_names
is_valid_lazyframe = getattr(
utils_mod, "is_valid_lazyframe", lambda df: isinstance(df, pl.LazyFrame)
)
get_row_count = getattr(
utils_mod,
"get_row_count",
lambda df: (
df.select(pl.len()).collect(engine="streaming")[0, 0]
if isinstance(df, pl.LazyFrame)
else df.height
),
)
logger = logging.getLogger(__name__)
def check_value_column_exist(
df: pl.DataFrame | pl.LazyFrame, valueCols: List[str]
) -> list[str]:
"""Return value columns present in ``df``."""
checkedValueCols: list[str] = []
columns, schema = get_schema_and_column_names(df)
for column in valueCols:
if column in columns:
checkedValueCols.append(column)
return checkedValueCols
def build_equal_to_query_string_element(colName: str, keyString: str) -> str:
"""Return a query fragment of the form ``col == 'value'``."""
leftSideElement = "" + colName + " == "
cleanedValue = str(keyString)
cleanedValue = cleanedValue.replace("'", "")
rightSideElement = "'" + cleanedValue + "'"
queryStringElement = leftSideElement + rightSideElement
return queryStringElement
def get_query_string_from_dict(filterDict: Dict[str, str]) -> str:
"""Build a boolean query string from a mapping of column to value."""
count = 0
fullQueryString = ""
for colName in filterDict:
queryStringElement = build_equal_to_query_string_element(
colName, filterDict[colName]
)
fullQueryString = assemble_query_string_elements(
queryStringElement, fullQueryString, count
)
count = count + 1
return fullQueryString
def take_price_out_of_valueCols(metricArray: List[str]) -> list[str]:
"""Return ``metricArray`` without any elements containing the price name."""
namingParams = get_naming_params()
priceName = namingParams["priceName"]
valueCols: list[str] = []
for element in metricArray:
if priceName not in element:
valueCols.append(element)
return valueCols
def assemble_query_string_elements(
queryStringElement: str, fullQueryString: str, count: int
) -> str:
"""Concatenate query fragments with ``and`` separators."""
if count == 0:
fullQueryString = queryStringElement
else:
fullQueryString = fullQueryString + " and " + queryStringElement
return fullQueryString
def complete_combination_with_all_parents(
testArray: list[str], indexDict: Dict[str, Any]
) -> list[str]:
"""Add missing hierarchical parents to ``testArray``."""
namingParams = get_naming_params()
hierarchicalName = namingParams["hierarchicalName"]
indexDict = indexDict[hierarchicalName]
for element in indexDict:
testArray = insert_missing_parents(testArray, indexDict[element])
testArray.sort()
return testArray
def delete_hierarchical_parents(
testArray: list[str], indexDict: Dict[str, Any]
) -> list[str]:
"""Remove duplicate hierarchy parents from ``testArray``."""
namingParams = get_naming_params()
hierarchicalName = namingParams["hierarchicalName"]
indexDict = indexDict[hierarchicalName]
for element in indexDict:
testArray = check_if_more_than_one_in_common(testArray, indexDict[element])
return testArray
def _reindex_by_period(df: pl.LazyFrame, period_order: list) -> pl.LazyFrame:
"""
Reindex a Polars LazyFrame by a specified period order.
"""
namingParams = get_naming_params()
periodName = namingParams["periodName"]
lookup = pl.DataFrame(
{periodName: period_order, "sort_order": range(len(period_order))}
).lazy()
return (
df.join(lookup, on=periodName, how="left").sort("sort_order").drop("sort_order")
)
def sort_periods_polars(
df: pl.LazyFrame, chartDict: dict, paramDict: dict
) -> pl.LazyFrame:
"""Sort periods using the historical ordering logic implemented with Polars."""
namingParams = get_naming_params()
yearName = namingParams["yearName"]
periodName = namingParams["periodName"]
periodChoice = namingParams["periodChoice"]
compareWithYearBefore = namingParams["compareWithYearBefore"]
selectedPeriods = namingParams["selectedPeriods"]
chosenChart = namingParams["chosenChart"]
stackedColumnChart = namingParams["stackedColumnChart"]
changedTimeAggregation = namingParams["changedTimeAggregation"]
chosenChart = chartDict[chosenChart]
periodChoice = chartDict[periodChoice]
periodOrder = chartDict[selectedPeriods]
df_count = df.select(pl.len()).collect(engine="streaming")[0, 0]
index_order = (
df.select(pl.col(periodName)).collect().get_column(periodName).to_list()
)
check_collect("ACA", "df_count", df_count)
check_collect("ADA", "indexOrder", index_order)
if (
periodChoice == yearName
and compareWithYearBefore in chartDict
and (
changedTimeAggregation not in paramDict
or not paramDict[changedTimeAggregation]
)
):
if df_count == 2:
df = _reindex_by_period(df, periodOrder)
elif chartDict[compareWithYearBefore]:
df = df.sort(periodName, descending=True)
else:
df = df.sort(periodName, descending=False)
elif index_order != periodOrder:
if set(index_order) == set(periodOrder):
df = _reindex_by_period(df, periodOrder)
else:
df = df.sort(periodName, descending=False)
return df
def transform_lazy_df(
lazy_df,
chartDict,
dimensionName,
itemName,
plotValuesAsChoice,
absolute,
percentName,
valueName,
):
"""
Applies the per-DataFrame transformations in lazy Polars:
1. Drop the existing valueName column if present
2. Create or ensure dimensionName is a string column
3. Melt either into (dimensionName, itemName, valueName) or
(dimensionName, itemName, percentName)
4. Compute the total-based percentage or multiply by the 'most recent'
absolute value if needed.
"""
columns, _ = get_schema_and_column_names(lazy_df) # Adapt this helper for Polars
# If the column to drop exists, drop it
if valueName in columns:
lazy_df = lazy_df.drop(valueName)
# In Pandas code, we reset the index named dimensionName into a column.
# Polars doesn't have an index concept, so adapt as needed:
# If dimensionName doesn't exist, you might do a row_count or similar.
if dimensionName not in columns:
lazy_df = lazy_df.with_row_index(name=dimensionName)
# Cast dimensionName to string
lazy_df = lazy_df.with_columns(pl.col(dimensionName).cast(pl.Utf8))
if plotValuesAsChoice in chartDict and chartDict[plotValuesAsChoice] == absolute:
lazy_cols, _ = get_schema_and_column_names(lazy_df)
cols_to_unpivot = [col for col in lazy_cols if col != dimensionName]
df_melted = lazy_df.unpivot(
index=[dimensionName],
on=cols_to_unpivot,
variable_name=itemName,
value_name=valueName,
)
# We need total of valueName to compute percentage:
total_df = df_melted.select(pl.col(valueName).sum().alias("totalValue"))
# Cross join total back in to avoid collecting:
df_melted = (
df_melted.join(total_df, how="cross")
.with_columns(
((pl.col(valueName) / pl.col("totalValue")) * 100)
.round(0)
.alias(percentName)
)
.drop("totalValue")
)
return df_melted
else:
lazy_cols, _ = get_schema_and_column_names(lazy_df)
cols_to_unpivot = [col for col in lazy_cols if col != dimensionName]
df_melted = lazy_df.unpivot(
index=[dimensionName],
on=cols_to_unpivot,
variable_name=itemName,
value_name=percentName,
)
# If "most recent" is from chartDict[absolute], we need its last row’s value:
# Assume chartDict[absolute] is also a lazy Polars frame that yields a single row in .tail(1).
df_most_recent = chartDict[absolute].tail(1)
# We must know which column in df_most_recent we want to multiply with.
# Suppose it has a single column called "most_recent_value".
# Or rename that column for clarity:
# e.g., rename the existing single column to "most_recent_value":
old_cols, _ = get_schema_and_column_names(df_most_recent)
if len(old_cols) == 1:
df_most_recent = df_most_recent.rename({old_cols[0]: "most_recent_value"})
else:
# If there's more than one column, adapt or select the right one
# Example: assume the first column is the numeric value:
df_most_recent = df_most_recent.rename(
{old_cols[0]: "most_recent_value"}
).select("most_recent_value")
# Cross join so every row can multiply the percent by the single "most_recent_value"
df_joined = df_melted.join(df_most_recent, how="cross")
df_joined = df_joined.with_columns(
(pl.col(percentName) * pl.col("most_recent_value")).alias(valueName)
).drop("most_recent_value")
return df_joined
def build_is_in_query_string_element(colName, array, operator):
"""
builds "must be equal to" query string
"""
leftSideElement = "" + colName + operator
cleanedValue = str(array)
cleanedValue = cleanedValue.replace("'", '"')
rightSideElement = "" + cleanedValue + ""
queryStringElement = leftSideElement + rightSideElement
return queryStringElement
def clean_array_values(array):
"""
replaces blances with underscores in array elements
"""
newArray = []
for element in array:
element = stripe_replace_and_clean(element, True)
newArray.append(element)
return newArray
def stripe_replace_and_clean(element, replacehyphen):
"""
to clean everything in the same way
"""
element = element.strip()
element = element.replace(",", "")
element = element.replace(".", "")
element = element.replace("'", "")
element = element.replace(" ", "_")
element = element.replace(":", "")
if replacehyphen:
element = element.replace("-", "_")
element = element.replace("/", "_")
return element
def filter_out_found_and_not_found_dataframe(
df_lazy: pl.LazyFrame, # equivalent to dfCopy in your original code
CXGRName: str,
foundArray: list,
frameArray: list,
chosenDimension: str, # unused in the snippet, but kept for signature compatibility
) -> tuple[pl.LazyFrame, pl.LazyFrame]:
"""
Split a LazyFrame using idiomatic Polars filters:
- ``dfFound``: rows where ``chosenDimension`` is in ``foundArray``
- ``dfNotFound``: rows where it is not, then drop ``CXGRName``
"""
# 1) Clone if you truly want a separate pipeline (mimicking 'duplicate_dataframe'):
# df_lazy = df_lazy.with_row_index("index_col")
df_clone = df_lazy.clone()
# 2) Filter rows where "index_col" is in foundArray
dfFound = df_clone.filter(pl.col(chosenDimension).is_in(foundArray))
# 3) Filter rows where "index_col" is NOT in foundArray
dfNotFound = df_clone.filter(~pl.col(chosenDimension).is_in(foundArray))
# 4) Drop the CXGRName column from dfNotFound
dfNotFound = dfNotFound.drop(CXGRName)
return dfNotFound, dfFound
def drop_AC_and_PY_month(df, column, valueCols, chartDict):
"""
if most recent date is at the middle of a month the resulting AC PY dataset can have a common/extra months that
makes calculations wrong and that we need to eliminate
"""
namingParams = get_naming_params()
periodName = namingParams["periodName"]
dateName = namingParams["dateName"]
acName = namingParams["acName"]
pyName = namingParams["pyName"]
plName = namingParams["plName"]
workColumn = namingParams["workColumn"]
workColumnTwo = namingParams["workColumnTwo"]
acPyArray = get_periods_array(df)
if plName not in acPyArray:
df = df.with_columns(pl.col(dateName).dt.strftime("%Y-%m").alias(workColumn))
group_byCols = [dateName, periodName, column]
for element in acPyArray:
if element != acName:
pyName = element
dfAc = df.filter(pl.col(periodName) == acName)
acDateArray = (
dfAc.select(pl.col(workColumn).unique()).collect()[workColumn].to_list()
)
dfPy = df.filter(pl.col(periodName) == pyName)
pyDateArray = (
dfPy.select(pl.col(workColumn).unique()).collect()[workColumn].to_list()
)
commonDates = set(acDateArray) & set(pyDateArray)
if commonDates:
df = df.with_columns(
pl.when(
(pl.col(workColumn).is_in(list(commonDates)))
& (pl.col(periodName) == acName)
)
.then(pyName)
.otherwise(pl.col(periodName))
.alias(periodName)
)
df = drop_columns(df, [workColumn])
valueCols = check_value_column_exist(df, valueCols)
df = df.group_by(group_byCols).agg([pl.col(col).sum() for col in valueCols])
df = df.with_columns(
pl.col(dateName).dt.strftime("%Y-%m").alias(workColumn)
)
dfAc = df.filter(pl.col(periodName) == acName)
acDateArray = (
dfAc.select(pl.col(workColumn).unique()).collect()[workColumn].to_list()
)
dfPy = df.filter(pl.col(periodName) == pyName)
pyDateArray = (
dfPy.select(pl.col(workColumn).unique()).collect()[workColumn].to_list()
)
pyDateArray.sort()
numberOfPyDates = len(pyDateArray)
if numberOfPyDates > 12:
dropDatesArray = pyDateArray[0 : numberOfPyDates - 12]
df = df.with_columns(
pl.when(
(pl.col(workColumn).is_in(dropDatesArray))
& (pl.col(periodName) == pyName)
)
.then(1)
.otherwise(None)
.alias(workColumnTwo)
)
df = df.filter(pl.col(workColumnTwo) != 1)
df = drop_columns(df, [workColumn, workColumnTwo])
return df
def get_number_of_multiples(df, mainDimension, chartDict):
"""
we compare the number of existing dimensions to those that the user wants to see. If too many, we return a smaller array of items
"""
namingParams = get_naming_params()
aggregateOtherWaterfalls = namingParams["aggregateOtherWaterfalls"]
aggregateOtherWaterfallsName = namingParams["aggregateOtherWaterfallsName"]
numberOfSmallMultiples = namingParams["numberOfSmallMultiplesWaterfall"]
nothingFilteredName = namingParams["nothingFilteredName"]
nothingThereString = namingParams["nothingThereString"]
numberOfSmallMultiples = chartDict[numberOfSmallMultiples]
aggregateOtherWaterfalls = chartDict[aggregateOtherWaterfalls]
# Use Polars to retrieve unique values, supporting LazyFrame/DataFrame
uniqueItems = (
ensure_lazyframe(df)
.select(pl.col(mainDimension).unique())
.collect()
.get_column(mainDimension)
.to_list()
)
showItems = []
possibleItems = []
count = 1
for element in uniqueItems:
if len(element) > 0 and element not in [
nothingFilteredName,
nothingThereString,
]:
possibleItems.append(element)
if count <= numberOfSmallMultiples:
showItems.append(element)
count = count + 1
if aggregateOtherWaterfalls and len(possibleItems) > len(showItems):
showItems.append(aggregateOtherWaterfallsName)
return showItems
def make_filtered_small_multiple_dataframe(
df, mainDimension, element, indexCols, chartDict, count
):
"""
we filter the dataframe on the item we need for a specific small multiple and take that item out of indexCols and of the dataframe
"""
df = aggregate_smaller_items(df, element, mainDimension, chartDict, count, False)
dfFiltered = df.filter(pl.col(mainDimension) == element)
df = df.filter(pl.col(mainDimension) != element)
dfFiltered = drop_columns(dfFiltered, [mainDimension])
if mainDimension in indexCols:
indexCols = take_filtered_value_out_of_option_list(indexCols, mainDimension)
return dfFiltered, df, indexCols
def aggregate_smaller_items(df, element, mainDimension, chartDict, count, filterDfBase):
"""
check if we need to aggregate all the remaining items
"""
namingParams = get_naming_params()
aggregateOtherWaterfalls = namingParams["aggregateOtherWaterfalls"]
aggregateOtherWaterfallsName = namingParams["aggregateOtherWaterfallsName"]
numberOfSmallMultiples = namingParams["numberOfSmallMultiplesWaterfall"]
nothingFilteredName = namingParams["nothingFilteredName"]
workColumn = namingParams["workColumn"]
varianceAmount = namingParams["varianceAmountName"]
varianceTypeName = namingParams["varianceTypeName"]
numberOfSmallMultiples = chartDict[numberOfSmallMultiples]
aggregateOtherWaterfalls = chartDict[aggregateOtherWaterfalls]
group_byCols = [varianceTypeName, mainDimension]
aggregationCols = [varianceAmount]
if (
aggregateOtherWaterfalls
and count == numberOfSmallMultiples + 1
and element == aggregateOtherWaterfallsName
):
df = df.with_columns(pl.lit(aggregateOtherWaterfallsName).alias(mainDimension))
if not filterDfBase:
group_byCols, aggregationCols = check_and_clean_columns(
df, group_byCols, aggregationCols
)
df = df.group_by(group_byCols).agg(
[pl.col(col).sum() for col in aggregationCols]
)
df = df.with_columns(pl.col(varianceAmount).abs().alias(workColumn))
df = df.sort(workColumn, descending=True)
df = add_running_total(df)
df = drop_columns(df, [workColumn])
return df
def get_subtotals(paramDict, chartDict, dfCopy, mainDimension, element, count, metric):
"""
we need to get the relevant initial and final subtotal for each small multiple plot
"""
namingParams = get_naming_params()
configParams = get_config_params()
multiplyConstant = configParams[namingParams["multiplyConstant"]]
isColumnMultiplied = namingParams["isColumnMultiplied"]
amountName = namingParams["monetaryLocalCurrencyName"]
netOfDiscount = namingParams["netOfDiscountName"]
marginName = namingParams["marginName"]
separatorString = namingParams["separatorString"]
periodsArray = configParams["periodsArray"]
varianceInPercent = namingParams["varianceInPercent"]
totalAmountPeriodZeroKey = namingParams["totalAmountPeriodZero"]
totalAmountPeriodOneKey = namingParams["totalAmountPeriodOne"]
isFilteredKey = namingParams["isFilteredKey"]
metConditionValue = namingParams["metConditionValue"]
notMetConditionValue = namingParams["notMetConditionValue"]
amountName = namingParams["monetaryLocalCurrencyName"]
discountName = namingParams["discountName"]
marginName = namingParams["marginName"]
df = aggregate_smaller_items(dfCopy, element, mainDimension, chartDict, count, True)
df = dfCopy.filter(pl.col(mainDimension) == element)
dfFiltered = dfCopy.filter(pl.col(mainDimension) != element)
columns, schema = get_schema_and_column_names(df)
amountPeriodZero = amountName + separatorString + periodsArray[0]
amountPeriodOne = amountName + separatorString + periodsArray[1]
totalPeriodZeroAmountNotFiltered = paramDict[totalAmountPeriodZeroKey]
totalPeriodOneAmountNotFiltered = paramDict[totalAmountPeriodOneKey]
if metric == amountName:
if amountPeriodZero in columns:
totalPeriodZeroValue = df.select(pl.col(amountPeriodZero).sum()).item()
totalPeriodOneValue = df.select(pl.col(amountPeriodOne).sum()).item()
if paramDict[isColumnMultiplied]:
totalPeriodZeroValue = totalPeriodZeroValue / multiplyConstant
totalPeriodOneValue = totalPeriodOneValue / multiplyConstant
elif metric == discountName:
netOfDiscountPeriodZero = netOfDiscount + separatorString + periodsArray[0]
netOfDiscountPeriodOne = netOfDiscount + separatorString + periodsArray[1]
if netOfDiscountPeriodZero in columns:
totalPeriodZeroValue = df.select(
pl.col(netOfDiscountPeriodZero).sum()
).item()
totalPeriodOneValue = df.select(pl.col(netOfDiscountPeriodOne).sum()).item()
if paramDict[isColumnMultiplied]:
totalPeriodZeroValue = totalPeriodZeroValue / multiplyConstant
totalPeriodOneValue = totalPeriodOneValue / multiplyConstant
if chartDict[varianceInPercent]:
totalPeriodZeroValue = (
totalPeriodZeroValue / totalPeriodZeroAmountNotFiltered * 100
)
totalPeriodOneValue = (
totalPeriodOneValue / totalPeriodOneAmountNotFiltered * 100
)
elif metric == marginName:
marginZero = marginName + separatorString + periodsArray[0]
marginOne = marginName + separatorString + periodsArray[1]
if marginZero in columns:
totalPeriodZeroValue = df.select(pl.col(marginZero).sum()).item()
totalPeriodOneValue = df.select(pl.col(marginOne).sum()).item()
if paramDict[isColumnMultiplied]:
totalPeriodZeroValue = totalPeriodZeroValue / multiplyConstant
totalPeriodOneValue = totalPeriodOneValue / multiplyConstant
if chartDict[varianceInPercent]:
totalPeriodZeroValue = (
totalPeriodZeroValue / totalPeriodZeroAmountNotFiltered * 100
)
totalPeriodOneValue = (
totalPeriodOneValue / totalPeriodOneAmountNotFiltered * 100
)
totalVarianceValue = totalPeriodOneValue - totalPeriodZeroValue
return totalPeriodZeroValue, totalPeriodOneValue, totalVarianceValue, dfFiltered
def drop_columns_with_all_blancs(df, outputIndexCols, toColorColumns, excludeArray):
"""
if a columns has all blancs in each row we need to drop it
"""
cleanedToColorColsArray = []
dropCols = []
for column in outputIndexCols:
if column not in excludeArray:
uniqueValues = (
ensure_lazyframe(df)
.select(pl.col(column).unique())
.collect()
.get_column(column)
.to_list()
)
if len(uniqueValues) == 1 and uniqueValues[0] == "":
dropCols.append(column)
for column in toColorColumns:
if column not in dropCols:
cleanedToColorColsArray.append(column)
df = drop_columns(df, dropCols)
return df, cleanedToColorColsArray
def add_missing_elements(
df: pl.DataFrame | pl.LazyFrame, elements: list[str]
) -> pl.LazyFrame:
"""Return ``df`` with all ``elements`` present and sorted by their order."""
namingParams = get_naming_params()
workColumn = namingParams["workColumn"]
lazy_df = df.lazy() if isinstance(df, pl.DataFrame) else df
lookup = (
pl.DataFrame({workColumn: elements}).with_row_index(name="sort_order").lazy()
)
missing_df = lookup.select(workColumn).join(
lazy_df.select(workColumn), on=workColumn, how="anti"
)
combined = pl.concat([lazy_df, missing_df], how="vertical")
result = (
combined.with_row_index(name="orig_order")
.join(lookup, on=workColumn, how="left")
.with_columns(pl.col("sort_order").fill_null(len(elements)))
.sort(["sort_order", "orig_order"])
.drop(["sort_order", "orig_order"])
)
return result
def concat_lazy_frames(frames: list[pl.LazyFrame]) -> pl.LazyFrame:
"""
Concatenate a list of Polars LazyFrames vertically and return
a single Polars LazyFrame.
"""
# If no frames are provided, return an empty LazyFrame
if not frames:
return pl.LazyFrame() # Optionally specify a schema if needed
return pl.concat(frames, how="vertical")
def concatenate_dfFound_and_dfNotFound_dataframes(
df: pl.LazyFrame, frameArray: list[pl.LazyFrame]
) -> pl.LazyFrame:
"""
If frameArray is non-empty, concatenate its contents (LazyFrames)
and return the result. Otherwise, just return df.
"""
if frameArray:
df = concat_lazy_frames(frameArray)
return df
def append_dfFound_and_dfNotFound_to_array(
dfFound: pl.LazyFrame, dfNotFound: pl.LazyFrame, element: int, frameArray: list
):
"""
Appends dfFound and dfNotFound to frameArray with a 'periodsMissing'
boolean column based on the provided naming parameters.
"""
namingParams = get_naming_params()
periodsMissing = namingParams["periodsMissing"]
# If element == 0, set dfFound[periodsMissing] = False
# and append it to frameArray
if element == 0:
dfFound = set_periods_missing_column(dfFound, periodsMissing, False)
frameArray.append(dfFound)
# For dfNotFound, set dfNotFound[periodsMissing] = True
# and append it to frameArray
dfNotFound = set_periods_missing_column(dfNotFound, periodsMissing, True)
frameArray.append(dfNotFound)
return frameArray, dfFound, dfNotFound
def set_CAGX_names(periodLengthInMonths, chartDict, paramDict):
namingParams = get_naming_params()
selectedPeriods = namingParams["selectedPeriods"]
periodChoice = namingParams["periodChoice"]
plName = namingParams["plName"]
deltaName = namingParams["deltaName"]
yearName = namingParams["yearName"]
quarterName = namingParams["quarterName"]
monthName = namingParams["monthName"]
weekName = namingParams["weekName"]
CAGRName = namingParams["CAGRName"]
CQGRName = namingParams["CQGRName"]
CMGRName = namingParams["CMGRName"]
CWGRName = namingParams["CWGRName"]
selectedPeriods = chartDict[selectedPeriods]
periodChoice = chartDict[periodChoice]
if selectedPeriods[0] == plName or not periodChoice:
CXGRName = deltaName
else:
periodDatesLengthInMonths = int(round(periodLengthInMonths, 0))
if periodChoice == yearName and periodDatesLengthInMonths >= 24:
CXGRName = CAGRName
elif periodChoice == yearName and periodDatesLengthInMonths <= 13:
CXGRName = CMGRName
elif periodChoice == yearName and periodDatesLengthInMonths < 24:
CXGRName = CQGRName
elif periodChoice == quarterName:
CXGRName = CQGRName
elif periodChoice == monthName:
CXGRName = CMGRName
elif periodChoice == weekName:
CXGRName = CWGRName
return CXGRName
def add_row_to_dataframe(
df: pl.DataFrame | pl.LazyFrame,
rowArray: list[Any],
endArray: list[Any],
position: str,
) -> pl.LazyFrame:
"""Append ``rowArray`` to ``df`` and return a ``LazyFrame``."""
df_lf = ensure_lazyframe(df)
columns, schema = get_schema_and_column_names(df_lf)
lengthDiff = len(columns) - (len(rowArray) + len(endArray))
if lengthDiff > 1:
rowArray = rowArray + [np.nan] * lengthDiff
rowArray = rowArray + endArray
new_row_df = pl.DataFrame([rowArray], schema=columns, orient="row")
if schema:
new_row_df = new_row_df.with_columns(
[
pl.col(column).cast(dtype, strict=False)
for column, dtype in schema.items()
if column in new_row_df.columns
]
)
new_row = new_row_df.lazy()
if position == "head":
return pl.concat([new_row, df_lf], how="vertical")
if position == "beforeLast":
insert_idx = get_row_count(df_lf) - 1
head = df_lf.slice(0, insert_idx)
tail = df_lf.slice(insert_idx, None)
return pl.concat([head, new_row, tail], how="vertical")
return pl.concat([df_lf, new_row], how="vertical")
def horizontal_cumsum(lf: pl.LazyFrame) -> pl.LazyFrame:
"""
Imitate axis=1 cumsum in Polars by iterating columns in order.
For each column:
running_sum = previous_sum + current_column
"""
columns, _ = get_schema_and_column_names(lf)
exprs = []
running_sum = pl.lit(0)
for col in columns:
running_sum = running_sum + pl.col(col)
exprs.append(running_sum.alias(col))
return lf.select(exprs)
def get_cum_sum_dataframe(
lf: pl.DataFrame | pl.LazyFrame,
chosenChart: str,
dfNegative,
message: str,
) -> pl.LazyFrame:
"""Return a cumulative-sum ``LazyFrame`` for chart rendering using Polars.
``lf`` may be either a :class:`~polars.DataFrame` or :class:`~polars.LazyFrame`.
"""
namingParams = get_naming_params()
valueName = namingParams["valueName"]
marimekkoChart = namingParams["marimekkoChart"]
lf_result = lf.lazy() if isinstance(lf, pl.DataFrame) else lf
# 1) Fill NaN with 0
lf_result = lf_result.fill_nan(0)
# 2) Select only numeric columns (float or integer types)
lf_result = lf_result.select(pl.selectors.numeric())
# 3) If chosenChart is "marimekko", drop the `valueName` column
if chosenChart == marimekkoChart:
lf_result = lf_result.drop(valueName)
# 4) Apply horizontal cumsum (simulating cumsum(axis=1))
lf_result = horizontal_cumsum(lf_result)
# 5) Handle dfNegative and message logic
# ``dfNegative`` can be a Polars frame or another sequence-like object.
# Use ``get_row_count`` when available to avoid ``len`` errors on LazyFrames.
try:
negative_count = (
get_row_count(dfNegative)
if isinstance(dfNegative, (pl.DataFrame, pl.LazyFrame))
else dfNegative.__len__()
)
except Exception as e:
logging.exception(e)
notifier.error("Something went wrong while computing the negative count.")
negative_count = 0
if negative_count == 0:
message = ""
return lf_result, dfNegative, message
def set_periods_missing_column(
df: pl.LazyFrame, column_name: str, value: bool
) -> pl.LazyFrame:
"""
Helper function to set a boolean column on a Polars LazyFrame.
"""
return df.with_columns(pl.lit(value).alias(column_name))
def concat_df_to_dfMerged(
df: pl.DataFrame | pl.LazyFrame, dfMerged: pl.DataFrame, dropCols
) -> pl.DataFrame:
"""Horizontally concatenate ``df`` onto ``dfMerged`` using Polars."""
df = ensure_polars_df(df).drop(dropCols)
return pl.concat([dfMerged, df], how="horizontal")
def sort_small_multiples(
df: pl.DataFrame | pl.LazyFrame, count: int, sortArray: list[str]
) -> tuple[pl.LazyFrame, list[str]]:
"""Return ``df`` lazily sorted by ``workColumn``."""
df_lf = ensure_lazyframe(df)
namingParams = get_naming_params()
workColumn = namingParams["workColumn"]
if count == 1:
sortArray = (
df_lf.select(pl.col(workColumn).unique())
.collect()
.get_column(workColumn)
.to_list()
)
else:
current = (
df_lf.select(pl.col(workColumn).unique())
.collect()
.get_column(workColumn)
.to_list()
)
sort_order = sortArray + [c for c in current if c not in sortArray]
df_lf = reindex_polars(df_lf, workColumn, sort_order)
return df_lf, sortArray
def calculate_total_cagr(
df_lazy: pl.LazyFrame, metric: str, CXGRName: str, periodOrder: list, count: int
) -> pl.LazyFrame:
"""
Polars-lazy equivalent pipeline:
1) Group by ``periodName`` and sum ``metric``
2) Add a literal ``totalName`` column
3) Compute CAGR with window expressions
4) Clean the CAGR column
"""
namingParams = get_naming_params()
periodName = namingParams["periodName"] # e.g. "period"
totalName = namingParams["totalName"] # e.g. "total"
# 1) Group by `periodName` and sum `metric`
df_lazy = df_lazy.group_by(periodName).agg(pl.col(metric).sum().alias(metric))
# 2) Add a column `totalName` with literal value = totalName
df_lazy = df_lazy.with_columns(pl.lit(totalName).alias(totalName))
# 3) Compute CAGR using a Polars-lazy window expression approach
df_lazy = compute_cagr_rate(
df_lazy=df_lazy,
CXGRName=CXGRName,
chosenDimension=totalName, # we group over the "total" dimension
periodOrder=periodOrder,
count=count,
metric_col=metric,
)
# 4) Clean the CAGR column (must also be a Polars-lazy function)
df_lazy = clean_cagr_column(df_lazy, CXGRName, periodOrder)
return df_lazy
def check_if_must_not_calculate_cagr(chartDict, paramDict):
namingParams = get_naming_params()
plotValuesAsChoice = namingParams["plotValuesAsChoice"]
absolute = namingParams["absolute"]
showCAGRKey = namingParams["showCAGR"]
notMetConditionValue = namingParams["notMetConditionValue"]
onePeriodOnly = namingParams["onePeriodOnly"]
showCAGR = notMetConditionValue
if chartDict[plotValuesAsChoice] != absolute:
chartDict[showCAGRKey] = notMetConditionValue
if onePeriodOnly in paramDict and paramDict[onePeriodOnly]:
chartDict[showCAGRKey] = notMetConditionValue
if showCAGRKey in chartDict:
showCAGR = chartDict[showCAGRKey]
return showCAGR
def add_count_metrics_to_cagr_array(CAGRArray, chartDict):
namingParams = get_naming_params()
countMetricsSumArrayKey = namingParams["countMetricsSumArray"]
countMetricsAvgArrayKey = namingParams["countMetricsAvgArray"]
if (
countMetricsSumArrayKey in chartDict
and len(chartDict[countMetricsSumArrayKey]) > 0
):
CAGRArray = CAGRArray + chartDict[countMetricsSumArrayKey]
if (
countMetricsAvgArrayKey in chartDict
and len(chartDict[countMetricsAvgArrayKey]) > 0
):
CAGRArray = CAGRArray + chartDict[countMetricsAvgArrayKey]
return CAGRArray
def sort_index_and_periods_for_cagr(
df_lazy: pl.LazyFrame,
chosenDimension: str,
xColumn: str,
metric: str,
chartDict: dict,
paramDict: dict,
) -> tuple[pl.LazyFrame, list]:
"""Return a tuple of ``(lazy_frame, periodOrder_list)`` computed with Polars."""
# 1) Fetch parameter dictionaries (like in your original code)
namingParams = get_naming_params()
configParams = get_config_params()
maxPeriodsForBarChart_key = namingParams["maxPeriodsForBarChart"]
periodChoice_key = namingParams["periodChoice"]
compareWithYearBefore_key = namingParams["compareWithYearBefore"]
yearName = namingParams["yearName"]
selectedPeriods_key = namingParams["selectedPeriods"]
plName = namingParams["plName"]
# 2) Extract dynamic values from chartDict and configParams
maxPeriodsForBarChart = configParams[maxPeriodsForBarChart_key]
periodChoice = chartDict[namingParams["periodChoice"]]
selectedPeriods = chartDict[namingParams["selectedPeriods"]]
# 3) Select only [chosenDimension, xColumn, metric]
df_lazy = df_lazy.select([chosenDimension, xColumn, metric])
# 4) Compute periodOrder lazily
if hasattr(utils_mod, "unique_list_lazy"):
periodOrder = utils_mod.unique_list_lazy(xColumn, df_lazy)
else:
periodOrder = (
df_lazy.select(pl.col(xColumn).unique())
.collect(engine="streaming")[xColumn]
.to_list()
)
check_collect("AAF", "periodOrder", periodOrder)
# 5) Decide sort order (ascending vs descending)
# This logic replicates:
# if periodChoice == yearName and compareWithYearBefore in chartDict and chartDict[compareWithYearBefore]:
# descending = True
# elif selectedPeriods[0] == plName:
# descending = True
# else:
# descending = False
if (
periodChoice == yearName
and compareWithYearBefore_key in chartDict
and chartDict[compareWithYearBefore_key]
):
descending = True
elif selectedPeriods and selectedPeriods[0] == plName:
descending = True
else:
descending = False
# 6) Sort by [chosenDimension, xColumn] in ascending or descending order
# This simulates "df.set_index([chosenDimension, xColumn])" + "df.sort_index()"
df_lazy = df_lazy.sort(by=[chosenDimension, xColumn], descending=descending)
# 7) Sort periodOrder the same way you did in Pandas
# (Note that in the original code, we only reversed periodOrder if descending)
periodOrder.sort(reverse=descending)
# 8) Slice the last N items from periodOrder
periodOrder = periodOrder[-maxPeriodsForBarChart:]
# 9) Return the LAZY frame and the pruned periodOrder list
return df_lazy, periodOrder
def pivot_cagr_dataframe(
df_lazy: pl.LazyFrame,
CXGRName: str,
xColumn: str,
chosenDimension: str,
agg_fn: str = "first", # or "sum", "mean", etc.
) -> pl.LazyFrame:
"""
Simulate a pivot in Polars lazy:
- index = xColumn
- columns = unique values in chosenDimension
- values = CXGRName
- aggregator = agg_fn (default "first")
"""
# 1) Collect unique values of `chosenDimension` into a Python list
dimension_values = (
df_lazy.select(pl.col(chosenDimension).unique())
.collect() # materialize to get the list
.get_column(chosenDimension)
.to_list()
)
check_collect("AAL", "dimension_values", dimension_values)
# 2) Build a list of aggregator expressions
# For each unique dimension value, we filter and apply the aggregator.
if agg_fn == "sum":
agg_exprs = [
pl.col(CXGRName).filter(pl.col(chosenDimension) == dv).sum().alias(str(dv))
for dv in dimension_values
]
elif agg_fn == "mean":
agg_exprs = [
pl.col(CXGRName).filter(pl.col(chosenDimension) == dv).mean().alias(str(dv))
for dv in dimension_values
]
else: # default "first"
agg_exprs = [
pl.col(CXGRName)
.filter(pl.col(chosenDimension) == dv)
.first()
.alias(str(dv))
for dv in dimension_values
]
# 3) Perform a group_by on xColumn and aggregate
df_pivoted_lazy = df_lazy.group_by(xColumn).agg(agg_exprs)
# Now `df_pivoted_lazy` is in wide form: one row per xColumn, columns for each dimension
return df_pivoted_lazy
def manage_missing_periods_for_cagr(
df: pl.LazyFrame, chosenDimension: str, xColumn: str, periodOrder: list[str]
) -> pl.LazyFrame:
"""Handle missing periods for CAGR calculations using Polars.
The pivot is performed lazily using :func:`pivot_lazy`.
``periodOrder`` is used only for debug logging via ``check_collect``.
"""
# Example external calls: adjust these to your actual code
namingParams = get_naming_params() # same as in your Pandas code
periodsMissing = namingParams["periodsMissing"] # column name for "periodsMissing"
columns, schema = get_schema_and_column_names(df)
# ^ your function that extracts column names, etc.
# Only do grouping/pivot if 'periodsMissing' is a known column
if periodsMissing in columns:
# Group by chosenDimension and xColumn, sum the periodsMissing
df_agg = df.group_by([chosenDimension, xColumn]).agg(
pl.col(periodsMissing).sum()
)
check_collect("AAAB", "missing periods", periodOrder)
# Perform the pivot lazily without collecting
df_pivot = pivot_lazy(
lf=df_agg,
index_col=xColumn,
pivot_col=chosenDimension,
value_col=periodsMissing,
agg_func="first",
)
return df_pivot
else:
# Return an empty Polars DataFrame if the column is missing
return pl.LazyFrame()
def load_cagr_data_into_chartdict(
dfCAGR, dfPeriodsMissing, chosenDimension, CXGRName, chartDict
):
namingParams = get_naming_params()
CXGRMetric = namingParams["CXGRMetricName"]
CXGRData = namingParams["CXGRData"]
CXGRTotal = namingParams["CXGRTotal"]
totalName = namingParams["totalName"]
periodsMissing = namingParams["periodsMissing"]
chartDict[CXGRMetric] = CXGRName
chartDict[periodsMissing] = dfPeriodsMissing
if chosenDimension == totalName:
chartDict[CXGRTotal] = dfCAGR
else:
chartDict[CXGRData] = dfCAGR
return chartDict
def calculate_cagr(dfCopy, chosenDimension, xColumn, metric, paramDict, chartDict):
"""
adds CAGR calculation
"""
namingParams = get_naming_params()
configParams = get_config_params()
CAGRArray = configParams[namingParams["CAGRArray"]]
showCAGRKey = namingParams["showCAGR"]
selectedPeriods = namingParams["selectedPeriods"]
totalName = namingParams["totalName"]
periodName = namingParams["periodName"]
CXGRMetric = namingParams["CXGRMetricName"]
periodsMissing = namingParams["periodsMissing"]
notMetConditionValue = namingParams["notMetConditionValue"]
selectedPeriods = chartDict[selectedPeriods]
frameArray = []
stop = False
showCAGR = check_if_must_not_calculate_cagr(chartDict, paramDict)
CAGRArray = add_count_metrics_to_cagr_array(CAGRArray, chartDict)
if metric in CAGRArray and is_valid_lazyframe(dfCopy) and showCAGR:
df = duplicate_dataframe(dfCopy)
paramDict, mostRecentDate, leastRecentDate, periodLengthInMonths = (
get_period_length(df, paramDict, False)
)
CXGRName = set_CAGX_names(periodLengthInMonths, chartDict, paramDict)
df, periodOrder = sort_index_and_periods_for_cagr(
df, chosenDimension, xColumn, metric, chartDict, paramDict
)
count = 1
if len(selectedPeriods) > 1:
dfCXGRTotal = pl.LazyFrame()
df = compute_cagr_rate(
df, CXGRName, chosenDimension, periodOrder, count, metric
)
if chosenDimension != totalName:
dfCAGR = duplicate_dataframe(df)
dfCXGRTotal = duplicate_dataframe(df)
dfCXGRTotal = calculate_total_cagr(
dfCXGRTotal, metric, CXGRName, periodOrder, count
)
for element in range(100):
if not stop and is_valid_lazyframe(dfCAGR):
count = count + 1
foundArray = extract_found_cagr_rates(
dfCAGR, CXGRName, chosenDimension
)
dfNotFound, dfFound = filter_out_found_and_not_found_dataframe(
dfCAGR, CXGRName, foundArray, frameArray, chosenDimension
)
numberOfPeriods = (
dfNotFound.select(
pl.col(periodName).n_unique()
) # expression for distinct count
.collect() # materialize
.item() # get the integer
)
check_collect("AAH", "numberOfPeriods", numberOfPeriods)
numberOfItems = (
dfNotFound.select(pl.col(chosenDimension).n_unique())
.collect()
.item()
)
check_collect("AAI", "numberOfItems", numberOfItems)
if (
is_valid_lazyframe(dfNotFound)
and numberOfPeriods > 1
and numberOfItems < get_row_count(dfNotFound)
):
cxgrPeriods = len(periodOrder) - count
dfNotFound = compute_cagr_rate(
dfNotFound,
CXGRName,
chosenDimension,
periodOrder,
count,
metric,
)
dfCAGR = duplicate_dataframe(dfNotFound)
frameArray, dfFound, dfNotFound = (
append_dfFound_and_dfNotFound_to_array(
dfFound, dfNotFound, element, frameArray
)
)
else:
stop = True
break
else:
pass
df = concatenate_dfFound_and_dfNotFound_dataframes(df, frameArray)
df = clean_cagr_column(df, CXGRName, periodOrder)
dfCAGR = pivot_cagr_dataframe(df, CXGRName, xColumn, chosenDimension)
dfPeriodsMissing = manage_missing_periods_for_cagr(
df, chosenDimension, xColumn, periodOrder
)
chartDict = load_cagr_data_into_chartdict(
dfCAGR, dfPeriodsMissing, chosenDimension, CXGRName, chartDict
)
if is_valid_lazyframe(dfCXGRTotal):
dfCXGRTotal = pivot_cagr_dataframe(
dfCXGRTotal, CXGRName, xColumn, totalName
)
chartDict = load_cagr_data_into_chartdict(
dfCXGRTotal, dfPeriodsMissing, totalName, CXGRName, chartDict
)
else:
chartDict[CXGRMetric] = notMetConditionValue
chartDict[showCAGRKey] = False
return chartDict, paramDict
def compute_cagr_rate(
df_lazy: pl.LazyFrame,
CXGRName: str,
chosenDimension: str,
periodOrder: list,
count: int,
metric_col: str,
) -> pl.LazyFrame:
"""
Compute CAGR per ``chosenDimension`` using Polars window expressions
(no pandas ``.apply``). Returns the LazyFrame with ``CXGRName`` appended.
- ``periodOrder`` and ``count`` determine the shift distance.
- ``metric_col`` is the numeric column used for CAGR.
"""
namingParams = get_naming_params()
periodsMissing = namingParams["periodsMissing"]
# 1) Calculate how many periods to shift for the CAGR
cxgrPeriods = len(periodOrder) - count
# 2) Drop 'periodsMissing' column if it exists
# df_lazy = df_lazy.drop(periodsMissing) # safe if column exists; else Polars may raise an error
# 3) Construct the window expression that mimics
# ((pct_change + 1) ** (1 / cxgrPeriods)) - 1 per group.
# NOTE: pct_change returns nulls for the first cxgrPeriods rows per group.
cagr_expr = (
(
(pl.col(metric_col).pct_change(n=cxgrPeriods).over(chosenDimension) + 1)
** (1 / cxgrPeriods)
)
- 1
).alias(CXGRName)
# 4) Add the new column to the DataFrame
df_lazy = df_lazy.with_columns(cagr_expr)
return df_lazy
def clean_cagr_column(
df: pl.LazyFrame, CXGRName: str, periodOrder: list
) -> pl.LazyFrame:
"""
1) Fill nulls with 0
2) Multiply by 100
3) Round to 1 decimal
4) If not the last period OR value == 0.0 => set null
5) Filter out null rows
6) Cast final values to string
"""
namingParams = get_naming_params()
periodName = namingParams["periodName"]
# Step 1-3: fill & round
df = df.with_columns((pl.col(CXGRName).fill_null(0) * 100).round(1).alias(CXGRName))
# Step 4: keep only the last period's non-zero values, else null
last_period = periodOrder[-1]
df = df.with_columns(
pl.when(
(pl.col(periodName).cast(pl.Utf8).str.strip_chars() == last_period)
& (pl.col(CXGRName) != 0)
)
.then(pl.col(CXGRName))
.otherwise(None)
.alias(CXGRName)
)
# Step 5: filter out null
df = df.filter(pl.col(CXGRName).is_not_null())
# Step 6: cast to string for final display if desired
df = df.with_columns(pl.col(CXGRName).cast(pl.Utf8).alias(CXGRName))
return df
def extract_found_cagr_rates(
df_lazy: pl.LazyFrame, CXGRName: str, chosenDimension: str
) -> list:
"""Return unique values from ``chosenDimension`` using Polars lazy steps.
Steps:
1) Duplicate the DataFrame if needed.
2) Filter rows where ``CXGRName`` is not null using ``filter``.
3) Select unique values of ``chosenDimension`` with ``unique``.
4) ``collect`` the result and convert the column to a list.
"""
# (Optional) If you truly need a second lazy pipeline to transform separately:
# df_lazy = df_lazy.clone()
# 1) Filter out rows where CXGRName is null
df_filtered = df_lazy.filter(pl.col(CXGRName).is_not_null())
# 2) Collect the unique values of chosenDimension into a Python list
foundArray = (
df_filtered.select(pl.col(chosenDimension).unique())
.collect() # materialize
.get_column(chosenDimension) # extract Series
.to_list() # convert to Python list
)
check_collect("AAG", "foundArray", foundArray)
return foundArray
def reindex_polars(
lf: pl.LazyFrame, xColumn: str, new_order_list: list
) -> pl.LazyFrame:
"""
Polars-lazy "reindex" equivalent:
1) build a small DataFrame with the desired order
2) join the main lazy frame on xColumn (how="left")
3) sort by that order's row count
4) drop the helper column
"""
order_df = (
pl.DataFrame({xColumn: new_order_list})
.with_row_index(name="temp_order", offset=0)
.lazy()
)
return (
order_df.join(lf, on=xColumn, how="left") # keep the order from order_df
.sort("temp_order")
.drop("temp_order")
)
def pivot_lazy(
lf: pl.LazyFrame | pl.DataFrame,
index_col: str,
pivot_col: str,
value_col: str,
agg_func: str = "sum",
) -> pl.LazyFrame:
"""Pivot ``lf`` on ``pivot_col`` using Polars.
Older versions of Polars do not expose ``LazyFrame.pivot``. In that case the
input is collected to a ``DataFrame`` and pivoted eagerly before returning a
lazy frame again.
"""
lf = ensure_lazyframe(lf)
def _pivot_df(df: pl.DataFrame, agg_key: str) -> pl.DataFrame:
agg_kwargs = {agg_key: agg_func}
attempts = [
(
(),
{
"columns": pivot_col,
"index": index_col,
"values": value_col,
**agg_kwargs,
},
),
(
(),
{
"on": pivot_col,
"index": index_col,
"values": value_col,
**agg_kwargs,
},
),
((pivot_col,), {"index": index_col, "values": value_col, **agg_kwargs}),
((pivot_col, index_col, value_col, agg_func), {}),
((pivot_col, index_col, value_col), agg_kwargs),
]
for args, kwargs in attempts:
try:
return df.pivot(*args, **kwargs)
except TypeError:
continue
return df.pivot(pivot_col, index_col, value_col, agg_func)
df = lf.collect()
try:
df_pivot = _pivot_df(df, "aggregate_function")
except TypeError:
df_pivot = _pivot_df(df, "aggregate_fn")
df_pivot = df_pivot.lazy()
columns, schema = get_schema_and_column_names(df_pivot)
rename_map = {c: f"{value_col}_{c}" for c in columns if c != index_col}
return df_pivot.rename(rename_map)
def clean_column_labels_after_flatten(
df: pl.LazyFrame, metricsToPlot: list[str]
) -> tuple[pl.LazyFrame, list]:
"""Remove metric prefixes from column labels produced by Polars pivots."""
# Example: rename columns that start with '<metric>_' for a single metric
columns, schema = get_schema_and_column_names(df)
# Just do a small rename if needed
rename_map = {}
for c in columns:
# e.g. remove "<metric>_" if single metric
for m in metricsToPlot:
prefix = f"{m}_"
if c.startswith(prefix) and len(metricsToPlot) == 1:
new_c = c.replace(prefix, "")
rename_map[c] = new_c
df = df.rename(rename_map) if rename_map else df
return df, columns
def clean_column_labels_after_flatten_polars(
df: pl.DataFrame | pl.LazyFrame, metricsToPlot
):
"""Remove metric prefixes from column names when flattened."""
columns, _ = get_schema_and_column_names(df)
newCols = []
for idx, column in enumerate(columns):
new_col = column
if idx != 0:
for metric in metricsToPlot:
new_col = new_col.replace(metric, "")
newCols.append(new_col)
rename_map = {old: new for old, new in zip(columns, newCols) if old != new}
df = df.rename(rename_map) if rename_map else df
return df, newCols
# Backwards compatibility alias
clean_column_labels_after_flatten_df = clean_column_labels_after_flatten_polars
def check_if_value_columns_in_df(df, metricArray):
columns, schema = get_schema_and_column_names(df)
valueCols = []
for element in metricArray:
if element in columns:
valueCols.append(element)
return valueCols
def drop_indirect_costs_rows(
df: pl.DataFrame | pl.LazyFrame, valueCols: list[str]
) -> pl.LazyFrame:
"""Remove rows where non-indirect metrics sum to ``<= 1``."""
namingParams = get_naming_params()
indirectCostsName = namingParams["indirectCostsName"]
workColumn = namingParams["workColumn"]
netMarginName = namingParams["netMarginName"]
indirectColsArray = [indirectCostsName, netMarginName]
valueColsWithoutIndirect = [
col for col in valueCols if col not in indirectColsArray
]
lf = ensure_lazyframe(df)
if indirectCostsName not in valueCols:
return lf
lf = lf.with_columns(
pl.sum_horizontal([pl.col(c) for c in valueColsWithoutIndirect]).alias(
workColumn
)
).filter(pl.col(workColumn) > 1)
lf = drop_columns(lf, [workColumn])
return lf
def get_unique_dimensions(df: pl.DataFrame | pl.LazyFrame, column: str) -> int:
"""Return the number of unique values for ``column`` without eager collect."""
lf = ensure_lazyframe(df)
number_of_unique_dimension = lf.select(pl.col(column).n_unique()).collect(
engine="streaming"
)[0, 0]
check_collect("AAN", "numberOfUniqueDimension", number_of_unique_dimension)
return int(number_of_unique_dimension)
def rank_dataframe(df: pl.DataFrame, column: str):
"""
Ranks the dataframe based on the sum of a monetary column, grouped by a specified column.
Args:
df (pl.DataFrame): The input dataframe.
column (str): The column to group by.
Returns:
Tuple[pl.DataFrame, bool]: The ranked dataframe with a rank column and a boolean flag.
"""
namingParams = get_naming_params()
monetaryName = namingParams["monetaryLocalCurrencyName"]
rank = namingParams["rankName"]
dfRank = duplicate_dataframe(
df
) # Ensure no modifications to the original dataframe
dfRank = (
dfRank.lazy()
.group_by(column)
.agg([pl.col(monetaryName).sum().alias(monetaryName)])
.with_columns(
pl.col(monetaryName).rank(method="average", descending=True).alias(rank)
)
.drop(monetaryName)
.sort(by=rank, descending=False)
)
sortByGlobalItems = True # This can be removed if not needed
return dfRank, sortByGlobalItems
def initialize_rank_boolean(df_lazy: pl.LazyFrame) -> pl.LazyFrame:
"""Initialize the ``rank_boolean`` column with ``not_met_condition_value``."""
namingParams = get_naming_params()
notMetConditionValue = namingParams["notMetConditionValue"]
rankBoolean = namingParams["rankBoolean"]
updated_df = df_lazy.with_columns(pl.lit(notMetConditionValue).alias(rankBoolean))
return updated_df
def scenario_top_condition(
df_lazy: pl.LazyFrame,
number_of_top: int,
) -> pl.LazyFrame:
"""
When rank <= number_of_top, set rank_boolean = met_condition_value,
otherwise set it to not_met_condition_value.
"""
# First pass: assign met_condition_value if rank <= number_of_top
namingParams = get_naming_params()
metConditionValue = namingParams["metConditionValue"]
notMetConditionValue = namingParams["notMetConditionValue"]
rankBoolean = namingParams["rankBoolean"]
rank = namingParams["rankName"]
updated_df = df_lazy.with_columns(
pl.when(pl.col(rank) <= number_of_top)
.then(metConditionValue)
.otherwise(pl.col(rankBoolean)) # leave existing value if condition not met
.alias(rankBoolean)
)
# Second pass: assign not_met_condition_value if not already met_condition_value
updated_df = updated_df.with_columns(
pl.when(pl.col(rankBoolean) != metConditionValue)
.then(notMetConditionValue)
.otherwise(pl.col(rankBoolean))
.alias(rankBoolean)
)
return updated_df
def scenario_global_items(
df_lazy: pl.LazyFrame,
column: str,
paramDict: dict,
) -> pl.LazyFrame:
"""
If column is in paramDict[globalUniqueItemsArrayKey],
set rank_boolean = met_condition_value,
else set it to not_met_condition_value.
"""
# First pass: set to met_condition_value if the column is in the global items array
namingParams = get_naming_params()
globalUniqueItemsArrayKey = namingParams["globalUniqueItemsArray"]
metConditionValue = namingParams["metConditionValue"]
notMetConditionValue = namingParams["notMetConditionValue"]
rankBoolean = namingParams["rankBoolean"]
rank = namingParams["rankName"]
updated_df = df_lazy.with_columns(
pl.when(pl.col(column).is_in(paramDict[globalUniqueItemsArrayKey]))
.then(metConditionValue)
.otherwise(pl.col(rankBoolean)) # leave existing value if condition not met
.alias(rankBoolean)
)
# Second pass: set to not_met_condition_value if not already met_condition_value
updated_df = updated_df.with_columns(
pl.when(pl.col(rankBoolean) != metConditionValue)
.then(notMetConditionValue)
.otherwise(pl.col(rankBoolean))
.alias(rankBoolean)
)
return updated_df
def scenario_filter_top(
df_lazy: pl.LazyFrame,
number_of_top: int,
) -> pl.LazyFrame:
"""
Filter rows where rank_col <= number_of_top,
then set rank_boolean = met_condition_value.
"""
namingParams = get_naming_params()
globalUniqueItemsArrayKey = namingParams["globalUniqueItemsArray"]
metConditionValue = namingParams["metConditionValue"]
notMetConditionValue = namingParams["notMetConditionValue"]
rankBoolean = namingParams["rankBoolean"]
rank = namingParams["rankName"]
filtered_df = df_lazy.filter(pl.col(rank) <= number_of_top)
updated_df = filtered_df.with_columns(pl.lit(metConditionValue).alias(rankBoolean))
return updated_df
def get_top_unique_items(df: pl.DataFrame, column: str, number_of_top: int) -> list:
"""
Returns the top N unique items in `df[column]`.
"""
# Create a lazy frame
# Select the column and get unique values
unique_lazy = df.select(pl.col(column)).unique(maintain_order=True)
# Collect as a DataFrame
unique_df = unique_lazy.collect()
check_collect("AAO", "unique_df", unique_df[:5])
# Convert to list and slice
unique_items = unique_df.get_column(column).to_list()[:number_of_top]
return unique_items
def process_df_rank(df, column, chartDict, paramDict, key):
namingParams = get_naming_params()
fatherAndChildDimensions = namingParams["fatherAndChildDimensions"]
showTopForEachItem = namingParams["showTopForEachItem"]
plotSmallMultiplesKey = namingParams["plotSmallMultiplesOtherCharts"]
globalUniqueItemsArrayKey = namingParams["globalUniqueItemsArray"]
marimekkoChart = namingParams["marimekkoChart"]
stackedBarChart = namingParams["stackedBarChart"]
chosenChart = namingParams["chosenChart"]
chosenChart = chartDict[chosenChart]
aggregateOtherItems = chartDict[key][namingParams["aggregateOtherItems"]]
numberOfTop = chartDict[key][namingParams["numberOfTop"]]
dfRank, sortByGlobalItems = rank_dataframe(df, column)
if (
fatherAndChildDimensions in chartDict and chartDict[fatherAndChildDimensions]
) or (showTopForEachItem in chartDict and chartDict[showTopForEachItem]):
sortByGlobalItems = False
dfRank = initialize_rank_boolean(dfRank)
if (
key == "W"
and chosenChart in [marimekkoChart, stackedBarChart]
and plotSmallMultiplesKey in chartDict
and chartDict[plotSmallMultiplesKey]
):
dfRank = scenario_top_condition(dfRank, numberOfTop)
elif (
globalUniqueItemsArrayKey in paramDict
and paramDict[globalUniqueItemsArrayKey]
and sortByGlobalItems
):
dfRank = scenario_global_items(dfRank, column, paramDict)
elif aggregateOtherItems:
dfRank = scenario_top_condition(dfRank, numberOfTop)
else:
dfRank = scenario_filter_top(dfRank, numberOfTop)
uniqueItems = get_top_unique_items(dfRank, column, numberOfTop)
return dfRank, uniqueItems
def tag_small_label_gaps(
dfCopy, element: str, column: str, chartDict: dict, minGap: float, filterPeriod: str
) -> pl.LazyFrame:
"""Tag labels that are too close to each other using Polars expressions."""
namingParams = get_naming_params()
periodName = namingParams["periodName"]
workColumn = namingParams["workColumn"]
workColumnTwo = namingParams["workColumnTwo"]
lf = ensure_lazyframe(dfCopy).select([column, element, periodName])
lf = lf.filter(pl.col(periodName) == filterPeriod).sort(element)
lf = (
lf.with_columns(
(pl.col(element) - pl.col(element).shift(-1)).abs().alias(workColumn)
)
.with_columns(
pl.when(pl.col(workColumn) <= minGap)
.then(0)
.otherwise(1)
.alias(workColumnTwo)
)
.drop([workColumn, element])
)
return lf
def add_yearly_average(
df: pl.DataFrame | pl.LazyFrame, y_array: list[str], chartDict: dict
) -> pl.LazyFrame:
"""Return ``df`` with an appended row of yearly averages as a ``LazyFrame``."""
namingParams = get_naming_params()
acName = namingParams["acName"]
plName = namingParams["plName"]
averageName = namingParams["averageName"]
dateName = namingParams["dateName"]
fcName = namingParams["fcName"]
compareScenariosOrPeriods = namingParams["compareScenariosOrPeriods"]
compareScenarios = namingParams["compareScenarios"]
lf = ensure_lazyframe(duplicate_dataframe(df))
if (
compareScenariosOrPeriods in chartDict
and chartDict[compareScenariosOrPeriods] == compareScenarios
and fcName in y_array
and plName in y_array
):
lf = lf.with_columns((pl.col(acName) + pl.col(fcName)).alias(fcName))
avg_row = (
lf.select([pl.col(col).mean().alias(col) for col in y_array])
.with_columns(pl.lit(averageName).alias(dateName))
.select([dateName] + y_array)
)
df_lf = ensure_lazyframe(df).with_columns(
[pl.col(col).cast(pl.Float64) for col in y_array]
)
empty_row = pl.LazyFrame({col: [None] for col in [dateName] + y_array})
result_lf = pl.concat([df_lf, avg_row, empty_row])
return result_lf
def rank_others_as_last(
df: pl.DataFrame | pl.LazyFrame, aggregate_other_items_name: str, rank_value: int
) -> pl.LazyFrame:
"""Rank rows containing ``aggregate_other_items_name`` at the end.
The function always returns a :class:`polars.LazyFrame` regardless of the
input type and never evaluates the lazy computation.
"""
namingParams = get_naming_params()
rank = namingParams["rankName"]
lf = ensure_lazyframe(df)
lf_cols, _ = get_schema_and_column_names(lf)
index_col = lf_cols[0]
label = pl.col(index_col).cast(pl.Utf8).str.to_lowercase()
is_other_label = (
label.eq("other")
| label.eq("other (aggregated)")
| label.str.starts_with("other ")
)
if aggregate_other_items_name:
is_other_label = is_other_label | label.str.contains(
aggregate_other_items_name.lower(), literal=True
)
lf = lf.with_row_index(rank).with_columns(
pl.when(is_other_label).then(rank_value).otherwise(pl.col(rank) + 1).alias(rank)
)
lf = lf.sort(rank).drop(rank)
return lf
def get_month_name(df):
"""
change month numbers to months three letter abbreviations
"""
namingParams = get_naming_params()
configParams = get_config_params()
dateName = namingParams["dateName"]
monthDict = configParams[namingParams["monthDict"]]
df = df.with_columns(
pl.col(dateName).dt.month().map_dict(monthDict).alias(dateName)
)
return df
def order_dataframe_by_month(
df: pl.DataFrame | pl.LazyFrame,
paramDict: dict,
indexSort: bool,
yArray: list[str],
) -> pl.DataFrame | pl.LazyFrame:
"""Reorder the rows of ``df`` according to calendar months.
The returned object preserves the input type (``DataFrame`` or ``LazyFrame``).
When the most recent month is not December the ordering is shifted so the
output always covers the latest 12 months in chronological order. A blank
row and an average row are appended when ``indexSort`` is ``False``.
"""
namingParams = get_naming_params()
configParams = get_config_params()
dateName = namingParams["dateName"]
averageName = namingParams["averageName"]
workColumn = namingParams["workColumn"]
monthDict = configParams[namingParams["monthDictInt"]]
paramDict, mostRecentDate, _, _ = get_period_length(df, paramDict, False)
mostRecentMonth = mostRecentDate.month
monthArray = list(monthDict.values())
if mostRecentMonth == 12:
reorderedArray = monthArray.copy()
else:
thisYearArray = monthArray[:mostRecentMonth]
lastYearArray = monthArray[mostRecentMonth:]
reorderedArray = lastYearArray + thisYearArray
is_lazy = isinstance(df, pl.LazyFrame)
lf = df if is_lazy else df.lazy()
lookup = pl.DataFrame(
{dateName: reorderedArray, "_sort": range(len(reorderedArray))}
).lazy()
if indexSort:
lf = lf.join(lookup, on=dateName, how="inner").sort("_sort").drop("_sort")
lf_cols, _ = get_schema_and_column_names(lf)
value_cols = [c for c in lf_cols if c != dateName]
lf = (
lf.with_columns(
pl.all_horizontal([pl.col(c).is_not_null() for c in value_cols]).alias(
"__keep__"
)
)
.filter(pl.col("__keep__"))
.drop("__keep__")
)
else:
reorderedArray += ["", averageName]
lookup = pl.DataFrame(
{dateName: reorderedArray, "_sort": range(len(reorderedArray))}
).lazy()
lf = lf.join(lookup, on=dateName, how="full")
lf = lf.with_columns(pl.col("_sort").fill_null(len(reorderedArray)))
cond = pl.all_horizontal([pl.col(col).is_null() for col in yArray]) & (
pl.col(dateName) != ""
)
lf = (
lf.sort("_sort")
.with_columns(pl.when(cond).then(1).otherwise(None).alias(workColumn))
.filter(pl.col(workColumn).is_null())
.drop(["_sort", workColumn])
)
return lf if is_lazy else lf.collect()
def identify_close_value_labels(
dfCopy: pl.DataFrame | pl.LazyFrame,
metric: str,
column: str,
paramDict: dict,
chartDict: dict,
) -> pl.LazyFrame:
"""Tag values that are too close to one another using Polars lazy APIs."""
namingParams = get_naming_params()
labelName = namingParams["labelName"]
workColumnTwo = namingParams["workColumnTwo"]
periodName = namingParams["periodName"]
periodChoice = namingParams["periodChoice"]
weekName = namingParams["weekName"]
quarterName = namingParams["quarterName"]
selectedPeriods = namingParams["selectedPeriods"]
periodOrder = chartDict[selectedPeriods]
lf = ensure_lazyframe(duplicate_dataframe(dfCopy)).with_columns(
pl.col(metric).alias(labelName)
)
max_value = ensure_lazyframe(dfCopy).select(pl.col(metric).max()).collect().item()
min_gap = max_value / 20
if periodChoice in chartDict and chartDict[periodChoice] in [weekName, quarterName]:
try:
lf = lf.with_columns(
pl.col(periodName).str.to_uppercase().alias(periodName)
)
except Exception as e: # noqa: BLE001 # nosec B110
logging.exception(e)
notifier.error("Something went wrong while uppercasing periods.")
lf, periodOrder[0] = check_if_periods_in_columns(lf, periodOrder[0])
dfPeriodZero = tag_small_label_gaps(
lf, metric, column, chartDict, min_gap, periodOrder[0]
)
lf, periodOrder[1] = check_if_periods_in_columns(lf, periodOrder[1])
dfPeriodOne = tag_small_label_gaps(
lf, metric, column, chartDict, min_gap, periodOrder[1]
)
dfGaps = pl.concat([dfPeriodZero, dfPeriodOne])
lf = (
lf.join(dfGaps, on=[column, periodName], how="left")
.sort([column, periodName])
.with_columns(
pl.when(pl.col(workColumnTwo) == 0)
.then(None)
.otherwise(pl.col(labelName))
.alias(labelName)
)
)
return lf
def compute_average_metrics(
df: pl.LazyFrame, chartDict: dict, countMetricsAvgDict: str, divisor_metric: str
) -> pl.LazyFrame:
"""
For each mapping ``new_col -> sum_col`` in ``chartDict[countMetricsAvgDict]``,
add ``new_col`` via ``with_columns((col(sum_col) / col(divisor_metric)).alias(new_col))``.
"""
# Nothing to do if countMetricsAvgDict is not in chartDict or is empty
if countMetricsAvgDict not in chartDict or len(chartDict[countMetricsAvgDict]) == 0:
return df
df_out = df
for new_col, sum_col in chartDict[countMetricsAvgDict].items():
df_out = df_out.with_columns(
(pl.col(sum_col) / pl.col(divisor_metric)).alias(new_col)
)
return df_out
def get_unique_counts(df, group_byCols, chartDict, countMetricsSumDict):
"""
Get unique counts for a given column and return the result as a lazy dataframe.
"""
namingParams = get_naming_params()
periodName = namingParams["periodName"]
dfCounts = pl.LazyFrame() # Default empty dataframe
if countMetricsSumDict in chartDict and len(chartDict[countMetricsSumDict]) > 0:
keyList = [*chartDict[countMetricsSumDict]]
colKey = keyList[0]
column = chartDict[countMetricsSumDict][colKey]
# Lazy operation to calculate unique counts
dfCounts = df.group_by(group_byCols).agg(
pl.col(column).n_unique().alias(colKey)
)
dfCounts = dfCounts.sort(
periodName
) # Assuming 'periodName' is the sorting criterion
return dfCounts
def adjust_group_by_columns(
chosenChart: str, group_byCols: list, yColumn: str, timeColumn: str, dateName: str
) -> list:
"""Adjust the columns used for group_by based on chart type and time column."""
namingParams = get_naming_params()
trendComparisonChart = namingParams["trendComparisonChart"]
multitierColumnChart = namingParams["multitierColumnChart"]
horizontalWaterfallChart = namingParams["horizontalWaterfallChart"]
timelineChart = namingParams["timelineChart"]
periodName = namingParams["periodName"]
if timeColumn == dateName:
if chosenChart in [
trendComparisonChart,
multitierColumnChart,
horizontalWaterfallChart,
]:
group_byCols = [periodName, yColumn, timeColumn]
elif chosenChart in [timelineChart]:
group_byCols = [yColumn, timeColumn]
else:
group_byCols = [yColumn, timeColumn]
return group_byCols
def get_number_of_uniques(dfCopy, yColumn, timeColumn, chartDict):
namingParams = get_naming_params()
countMetricsSumDict = namingParams["countMetricsSumDict"]
countMetricsSumArray = namingParams["countMetricsSumArray"]
countMetricsAvgArray = namingParams["countMetricsAvgArray"]
periodName = namingParams["periodName"]
dateName = namingParams["dateName"]
trendComparisonChart = namingParams["trendComparisonChart"]
multitierColumnChart = namingParams["multitierColumnChart"]
horizontalWaterfallChart = namingParams["horizontalWaterfallChart"]
timelineChart = namingParams["timelineChart"]
areaChart = namingParams["areaChart"]
chosenChart = namingParams["chosenChart"]
chosenChart = chartDict[chosenChart]
df = duplicate_dataframe(
dfCopy
) # Assuming this function makes the dataframe copy correctly
group_byCols = [periodName, yColumn]
# Adjust the group_byCols based on the timeColumn and chart
group_byCols = adjust_group_by_columns(
chosenChart, group_byCols, yColumn, timeColumn, dateName
)
dfCounts = get_unique_counts(df, group_byCols, chartDict, countMetricsSumDict)
return dfCounts, chartDict
def fill_other_items_for_metric(
df: pl.LazyFrame,
dfCounts: pl.LazyFrame,
metric: str,
yColumn: str,
aggregateOtherItemsName: str,
groupCols: list,
) -> pl.LazyFrame:
"""
Given df and dfCounts (both lazy), compute the total for `metric` from dfCounts
grouped by groupCols, and the sum of `metric` from df (excluding the 'Other' row),
then fill the 'Other' row with (total - sum_of_non_other).
"""
# Sum of counts for every group
total_df = dfCounts.group_by(groupCols).agg(
pl.col(metric).sum().alias("total_metric")
)
# Sum of metric for all non-"Other" rows in df
sum_non_others_df = (
df.filter(pl.col(yColumn) != aggregateOtherItemsName)
.group_by(groupCols)
.agg(pl.col(metric).sum().alias("joined_metric"))
)
# Join those aggregates back onto df so we can compute (total - joined)
df_out = (
df
# Bring in the sum of non-others
.join(sum_non_others_df, on=groupCols, how="left")
# Bring in the total from dfCounts
.join(total_df, on=groupCols, how="left")
# Conditionally fill the "Other" row with (total - joined), else keep original metric
.with_columns(
pl.when(pl.col(yColumn) == aggregateOtherItemsName)
.then(pl.col("total_metric") - pl.col("joined_metric"))
.otherwise(pl.col(metric))
.alias(metric)
)
# Clean up any possible nulls if needed, or do fill_null(0) – depends on your use case
# .with_column(pl.col(metric).fill_null(0))
.drop(["joined_metric", "total_metric"]) # remove helper columns
)
return df_out
def join_unique_metric_to_df(
df: pl.LazyFrame, # already Lazy
dfCounts: pl.LazyFrame, # assume also Lazy
yColumn: str,
timeColumn: str,
aggregateOtherItemsName: str,
chartDict: dict,
) -> pl.LazyFrame:
"""Join metrics lazily using Polars.
This relies on ``get_naming_params()`` to obtain column names and mappings.
"""
namingParams = get_naming_params()
periodName = namingParams["periodName"]
dateName = namingParams["dateName"]
totalName = namingParams["totalName"]
countMetricsAvgDict = namingParams["countMetricsAvgDict"]
countMetricsSumDict = namingParams["countMetricsSumDict"]
# If dfCounts is None or truly empty, just return df
# (We cannot lazily check the number of rows without collecting,
# so we skip that or do a "dfCounts is not None" check.)
if dfCounts is None:
return df
# If there are no sum metrics to process, just return df
if countMetricsSumDict not in chartDict or len(chartDict[countMetricsSumDict]) == 0:
return df
# Figure out which columns exist
df_columns, _ = get_schema_and_column_names(df)
group_byCols = [periodName, yColumn] # default
# If timeColumn == dateName, adapt group_byCols
if timeColumn == dateName:
if periodName in df_columns:
group_byCols = [periodName, yColumn, timeColumn]
else:
group_byCols = [yColumn, timeColumn]
# Remove duplicates in case periodName == yColumn or similar
group_byCols = list(set(group_byCols))
# 1) Join df and dfCounts on group_byCols so we have the columns from dfCounts
# (In your original code, you did a separate join per-metric, but we can
# do a single join once, because Polars can carry all columns.)
df_joined = df.join(dfCounts, on=group_byCols, how="left")
# 2) For each SUM metric, fill the "Other" row with (total - sum_of_non_other).
# Only do that if yColumn != totalName, matching your original condition.
df_out = df_joined
if totalName not in [None, ""]:
# If the grouping column is not the "total" dimension, handle the difference logic
df_columns_joined, _ = get_schema_and_column_names(df_joined) # after the join
if yColumn != totalName:
for metric in chartDict[countMetricsSumDict]:
# Only fill other items if metric is actually in dfCounts (joined)
if metric in df_columns_joined:
# fill "Other" row difference
# We pass group_byCols that do *not* include yColumn in the grouping
# if we want the sum of all categories except "Other."
# However, your original code lumps yColumn into the index.
# We'll replicate your logic with a separate function:
# fill_other_items_for_metric(…)
#
# If your original pivot logic expects grouping only by [periodName]
# or [dateName], you can adapt. For now, we use the same group_byCols
# (minus yColumn check inside the function).
df_out = fill_other_items_for_metric(
df=df_out,
dfCounts=dfCounts,
metric=metric,
yColumn=yColumn,
aggregateOtherItemsName=aggregateOtherItemsName,
groupCols=[
c for c in group_byCols if c != yColumn
], # replicate "all except the actual y cat"
)
# 3) If we have average metrics, compute them
# df[element] = df[ chartDict[countMetricsAvgDict][element] ] / df[ metric ]
# Notice that in your code, you used 'metric' from the sum loop, but effectively
# you do a second loop. We'll do a simpler approach: if there's anything
# in countMetricsAvgDict, you presumably want to divide by each sum metric.
# If your logic always divides by the *same* metric, adapt as needed.
if (countMetricsAvgDict in chartDict) and len(chartDict[countMetricsAvgDict]) > 0:
# Your original code suggests we do the division using the last 'metric' in the sum loop
# or all sum metrics. That part is ambiguous. We'll assume you want it per sum metric.
# If you truly only want "the sum metric" from the sum loop, pick the relevant one.
# We'll demonstrate dividing by the *first* sum metric (like your code).
first_sum_metric = list(chartDict[countMetricsSumDict])[0]
df_out = compute_average_metrics(
df_out, chartDict, countMetricsAvgDict, first_sum_metric
)
return df_out
def sort_by_column(lazy_df: pl.LazyFrame, column: str) -> pl.LazyFrame:
"""
Returns a new LazyFrame sorted by `column`.
"""
return lazy_df.sort(column)
def left_join_on_column(
lazy_left: pl.LazyFrame, lazy_right: pl.LazyFrame, on: str
) -> pl.LazyFrame:
"""
Left-joins two Polars LazyFrames on `on`.
"""
return lazy_left.join(lazy_right, on=on, how="left")
def sort_and_left_join(
df: pl.LazyFrame,
df_rank: pl.LazyFrame,
column: str,
) -> (pl.LazyFrame, list):
"""Sort two frames, then left join them in Polars lazy mode."""
# 2) Sort df_rank by column
lazy_df_rank = sort_by_column(df_rank, column)
# 3) Sort df by column
lazy_df = sort_by_column(df, column)
# 4) Left join
joined = left_join_on_column(lazy_df, lazy_df_rank, column)
# 5) Optional: mimic reset_index by adding a row count.
# If you don't want this, just leave it out.
# joined = joined.with_row_index(name="index")
return joined
def process_dataframe_based_on_chart_more_unique_dimensions(
df,
column,
secondColumn,
timeColumn,
valueCols,
uniqueItems,
aggregateOtherItemsName,
chartDict,
):
namingParams = get_naming_params()
rankBoolean = namingParams["rankBoolean"]
metConditionValue = namingParams["metConditionValue"]
notMetConditionValue = namingParams["notMetConditionValue"]
rank = namingParams["rankName"]
timelineChart = namingParams["timelineChart"]
dotChart = namingParams["dotChart"]
slopeChart = namingParams["slopeChart"]
vennChart = namingParams["vennChart"]
upsetChart = namingParams["upsetChart"]
multitierBarChart = namingParams["multitierBarChart"]
multitierColumnChart = namingParams["multitierColumnChart"]
horizontalWaterfallChart = namingParams["horizontalWaterfallChart"]
plotSmallMultiplesKey = namingParams["plotSmallMultiplesOtherCharts"]
areaChart = namingParams["areaChart"]
kernelDensity = namingParams["kernelDensityChart"]
histogramChart = namingParams["histogramChart"]
boxplotChart = namingParams["boxplotChart"]
stripplotChart = namingParams["stripplotChart"]
ecdfChart = namingParams["ecdfChart"]
bubbleChart = namingParams["bubbleChart"]
motionChart = namingParams["motionChart"]
scatterChart = namingParams["scatterChart"]
trendComparisonByPeriodChart = namingParams["trendComparisonByPeriodChart"]
trendComparisonChart = namingParams["trendComparisonChart"]
marimekkoChart = namingParams["marimekkoChart"]
barmekkoChart = namingParams["barmekkoChart"]
stackedBarChart = namingParams["stackedBarChart"]
periodName = namingParams["periodName"]
acpyName = namingParams["acpyName"]
chosenChart = namingParams["chosenChart"]
chosenChart = chartDict[chosenChart]
yAxisDimension = namingParams["yAxisDimension"]
df = df.with_columns(
pl.when(pl.col(rankBoolean) != metConditionValue)
.then(pl.lit(aggregateOtherItemsName))
.otherwise(pl.col(column))
.alias(column)
)
uniqueItems.append(aggregateOtherItemsName)
df = drop_columns(df, [rank])
columns, schema = get_schema_and_column_names(df)
if chosenChart in [
trendComparisonChart,
multitierColumnChart,
horizontalWaterfallChart,
]:
group_byCols = [column, timeColumn, periodName]
elif chosenChart in [timelineChart, areaChart]:
group_byCols = [column, timeColumn]
elif chosenChart in [trendComparisonByPeriodChart]:
group_byCols = [column, timeColumn, acpyName]
elif chosenChart in [
marimekkoChart,
vennChart,
upsetChart,
stackedBarChart,
bubbleChart,
]:
group_byCols = []
for element in [column, timeColumn, secondColumn]:
if element in columns and element not in group_byCols:
group_byCols.append(element)
if (
chosenChart in [upsetChart, vennChart]
and plotSmallMultiplesKey in chartDict
and chartDict[plotSmallMultiplesKey]
):
if yAxisDimension in chartDict:
yAxisDimension = chartDict[yAxisDimension]
if yAxisDimension in columns and yAxisDimension not in group_byCols:
group_byCols.append(yAxisDimension)
elif chosenChart in [barmekkoChart]:
group_byCols = [column, timeColumn]
elif chosenChart not in [
scatterChart,
kernelDensity,
histogramChart,
boxplotChart,
stripplotChart,
ecdfChart,
]:
group_byCols = [column, timeColumn]
else:
group_byCols = [column, timeColumn]
df, group_byCols, valueCols = check_and_group_by_cols(df, group_byCols, valueCols)
return df, group_byCols, valueCols, uniqueItems
def process_dataframe_based_on_chart_less_unique_dimensions(
df, column, secondColumn, timeColumn, valueCols, uniqueItems, chartDict
):
namingParams = get_naming_params()
timelineChart = namingParams["timelineChart"]
dotChart = namingParams["dotChart"]
slopeChart = namingParams["slopeChart"]
vennChart = namingParams["vennChart"]
upsetChart = namingParams["upsetChart"]
multitierBarChart = namingParams["multitierBarChart"]
multitierColumnChart = namingParams["multitierColumnChart"]
horizontalWaterfallChart = namingParams["horizontalWaterfallChart"]
areaChart = namingParams["areaChart"]
kernelDensity = namingParams["kernelDensityChart"]
histogramChart = namingParams["histogramChart"]
boxplotChart = namingParams["boxplotChart"]
stripplotChart = namingParams["stripplotChart"]
ecdfChart = namingParams["ecdfChart"]
bubbleChart = namingParams["bubbleChart"]
motionChart = namingParams["motionChart"]
scatterChart = namingParams["scatterChart"]
trendComparisonByPeriodChart = namingParams["trendComparisonByPeriodChart"]
trendComparisonChart = namingParams["trendComparisonChart"]
marimekkoChart = namingParams["marimekkoChart"]
barmekkoChart = namingParams["barmekkoChart"]
stackedBarChart = namingParams["stackedBarChart"]
periodName = namingParams["periodName"]
acpyName = namingParams["acpyName"]
chosenChart = namingParams["chosenChart"]
chosenChart = chartDict[chosenChart]
yAxisDimension = namingParams["yAxisDimension"]
nothingFilteredName = namingParams["nothingFilteredName"]
plotSmallMultiplesKey = namingParams["plotSmallMultiplesOtherCharts"]
columns, schema = get_schema_and_column_names(df)
if chosenChart in [
trendComparisonChart,
multitierColumnChart,
horizontalWaterfallChart,
]:
group_byCols = [column, timeColumn, periodName]
elif chosenChart in [timelineChart, areaChart]:
group_byCols = [column, timeColumn]
elif chosenChart in [trendComparisonByPeriodChart]:
group_byCols = [column, timeColumn, acpyName]
elif chosenChart in [marimekkoChart, vennChart, upsetChart, stackedBarChart]:
if secondColumn not in [nothingFilteredName] and secondColumn != None:
group_byCols = [column, timeColumn, secondColumn]
if (
chosenChart in [upsetChart, vennChart]
and plotSmallMultiplesKey in chartDict
and chartDict[plotSmallMultiplesKey]
):
if yAxisDimension in chartDict:
yAxisDimension = chartDict[yAxisDimension]
if yAxisDimension in columns and yAxisDimension not in group_byCols:
group_byCols.append(yAxisDimension)
else:
group_byCols = [column, timeColumn]
elif chosenChart in [barmekkoChart]:
group_byCols = [column, timeColumn]
elif chosenChart not in [
scatterChart,
kernelDensity,
histogramChart,
boxplotChart,
stripplotChart,
ecdfChart,
]:
group_byCols = [column, timeColumn]
else:
group_byCols = [column, timeColumn]
df, group_byCols, valueCols = check_and_group_by_cols(df, group_byCols, valueCols)
df = df.sort(
by=valueCols[0], # the column to sort by
descending=True, # ascending=False in Pandas => reverse=True in Polars
)
uniqueItems = (
df.select( # ensure we're in lazy mode
pl.col(column).unique()
) # pick the column and get its unique values
.collect()[ # materialize the lazy query
column
] # extract the column from the DataFrame
.to_list() # convert to a Python list
)
check_collect("AAM", "uniqueItems", uniqueItems)
return df, group_byCols, valueCols, uniqueItems
def show_only_largest(
dfCopy, column, secondColumn, timeColumn, valueColsCopy, chartDict, paramDict, key
):
"""
we just want to keep the top items
"""
namingParams = get_naming_params()
metConditionValue = namingParams["metConditionValue"]
notMetConditionValue = namingParams["notMetConditionValue"]
rank = namingParams["rankName"]
numberOfTop = chartDict[key][namingParams["numberOfTop"]]
aggregateOtherItems = chartDict[key][namingParams["aggregateOtherItems"]]
aggregateOtherItemsNameKey = namingParams["aggregateOtherItemsName"]
nothingThereString = namingParams["nothingThereString"]
fatherAndChildDimensions = namingParams["fatherAndChildDimensions"]
showTopForEachItem = namingParams["showTopForEachItem"]
WnumberOfTop = namingParams["WnumberOfTop"]
cleanedUniqueItems = []
uniqueItems = []
aggregateOtherItemsName = aggregateOtherItemsNameKey + str(numberOfTop)
df = duplicate_dataframe(dfCopy)
valueColsCopy = check_if_value_columns_in_df(df, valueColsCopy)
valueCols = take_price_out_of_valueCols(valueColsCopy)
monetaryName = namingParams["monetaryLocalCurrencyName"]
columns, _ = get_schema_and_column_names(df)
if monetaryName not in columns and valueCols:
df = df.with_columns(pl.col(valueCols[0]).alias(monetaryName))
df = drop_indirect_costs_rows(df, valueCols)
numberOfUniqueDimension = get_unique_dimensions(df, column)
if numberOfUniqueDimension > numberOfTop:
dfRank, uniqueItems = process_df_rank(df, column, chartDict, paramDict, key)
df = sort_and_left_join(df, dfRank, column)
if aggregateOtherItems:
df, group_byCols, valueCols, uniqueItems = (
process_dataframe_based_on_chart_more_unique_dimensions(
df,
column,
secondColumn,
timeColumn,
valueCols,
uniqueItems,
aggregateOtherItemsName,
chartDict,
)
)
else:
df = df.filter(pl.col(rank) == metConditionValue)
df = drop_columns(df, [rank])
else:
df, group_byCols, valueCols, uniqueItems = (
process_dataframe_based_on_chart_less_unique_dimensions(
df, column, secondColumn, timeColumn, valueCols, uniqueItems, chartDict
)
)
for element in uniqueItems:
if element != nothingThereString:
cleanedUniqueItems.append(element)
return df, cleanedUniqueItems, aggregateOtherItemsName, valueCols
def calculate_year_to_year_change(df, growthMetric, periodOrder):
df = df.with_columns(pl.lit(None).alias(growthMetric))
columns, schema = get_schema_and_column_names(df)
checkedPeriods = []
for period in periodOrder:
if period not in columns:
period = period.lower()
checkedPeriods.append(period)
columns, schema = get_schema_and_column_names(df)
if checkedPeriods[0] in columns and checkedPeriods[1] in columns:
df = df.with_columns(
pl.when(pl.col(checkedPeriods[0]) != 0)
.then(
(
(pl.col(checkedPeriods[1]) - pl.col(checkedPeriods[0]))
/ pl.col(checkedPeriods[0])
* 100
).round(0)
)
.otherwise(None)
.alias(growthMetric)
)
df = df.with_columns(
pl.when((pl.col(checkedPeriods[1]) < 0) & (pl.col(checkedPeriods[0]) < 0))
.then(pl.col(growthMetric) * -1)
.otherwise(pl.col(growthMetric))
.alias(growthMetric)
)
df = df.with_columns(
pl.when(pl.col(growthMetric) > 999)
.then(999)
.otherwise(pl.col(growthMetric))
.alias(growthMetric)
)
else:
df = df.with_columns(pl.lit(None).alias(growthMetric))
return df, checkedPeriods
def insert_unit_and_volume_price_column(df_lazy: pl.LazyFrame) -> pl.LazyFrame:
"""Add pricing columns using Polars' lazy API."""
# --- 1) Get naming parameters (same as in your code) ---
naming_params = get_naming_params()
units_name = naming_params["unitsName"]
price_per_unit_name = naming_params["pricePerUnitName"]
volume_name = naming_params["volumeName"]
price_per_volume_name = naming_params["pricePerVolumeName"]
amount_name = naming_params["monetaryLocalCurrencyName"]
discount_name = naming_params["discountName"]
net_of_discount_name = naming_params["netOfDiscountName"]
price_per_unit_net_discount_name = naming_params["pricePerUnitNetDiscountName"]
price_per_volume_net_discount_name = naming_params["pricePerVolumeNetDiscountName"]
discount_in_percent_name = naming_params["discountInPercentName"]
margin_in_percent_name = naming_params["marginInPercentName"]
margin_in_percent_of_net_sales_name = naming_params["marginInPercentOfNetSalesName"]
margin_name = naming_params["marginName"]
# --- 2) Get the column names from the LazyFrame schema ---
columns, schema = get_schema_and_column_names(df_lazy)
existing_columns = set(schema.keys()) # or df_lazy.columns for Polars >= 0.17.4
# We'll build a list of Polars expressions applied in a single ``with_columns`` call.
# Each block implements the conditional logic from the original pandas version.
new_columns = []
# ============== BLOCK A ==============
# When both unit and amount columns exist, use Polars to compute
# ``pricePerUnitName`` as ``amountName / unitsName`` only when
# ``unitsName`` is non-zero.
if {units_name, amount_name}.issubset(existing_columns):
expr_price_per_unit = (
pl.when(pl.col(units_name) != 0)
.then(pl.col(amount_name) / pl.col(units_name))
.otherwise(0)
.alias(price_per_unit_name)
.cast(pl.Float64)
)
new_columns.append(expr_price_per_unit)
# ============== BLOCK B ==============
# With ``volumeName`` and ``amountName`` present, compute
# ``pricePerVolumeName`` using ``amountName / volumeName`` when
# ``volumeName`` is non-zero.
if {volume_name, amount_name}.issubset(existing_columns):
expr_price_per_volume = (
pl.when(pl.col(volume_name) != 0)
.then(pl.col(amount_name) / pl.col(volume_name))
.otherwise(0)
.alias(price_per_volume_name)
.cast(pl.Float64)
)
new_columns.append(expr_price_per_volume)
# ============== BLOCK C ==============
# Using Polars expressions, combine ``unitsName``, ``discountName`` and
# ``netOfDiscountName`` to create ``pricePerUnitNetDiscountName`` as
# ``netOfDiscountName / unitsName`` when ``unitsName`` is non-zero and
# compute ``discountInPercentName`` from ``discountName / amountName`` when
# ``amountName`` is non-zero.
if {units_name, discount_name, net_of_discount_name}.issubset(existing_columns):
expr_price_per_unit_net = (
pl.when(pl.col(units_name) != 0)
.then(pl.col(net_of_discount_name) / pl.col(units_name))
.otherwise(0)
.alias(price_per_unit_net_discount_name)
.cast(pl.Float64)
)
expr_discount_percent = (
pl.when(pl.col(amount_name) != 0)
.then(pl.col(discount_name) / pl.col(amount_name))
.otherwise(0)
.alias(discount_in_percent_name)
.cast(pl.Float64)
)
new_columns.extend([expr_price_per_unit_net, expr_discount_percent])
# ============== BLOCK D ==============
# Using Polars, derive ``pricePerVolumeNetDiscountName`` as
# ``netOfDiscountName / volumeName`` when ``volumeName`` is non-zero and
# compute ``discountInPercentName`` from ``discountName / amountName``
# when ``amountName`` is non-zero.
if {volume_name, discount_name, net_of_discount_name}.issubset(existing_columns):
expr_price_per_volume_net = (
pl.when(pl.col(volume_name) != 0)
.then(pl.col(net_of_discount_name) / pl.col(volume_name))
.otherwise(0)
.alias(price_per_volume_net_discount_name)
.cast(pl.Float64)
)
expr_discount_percent_2 = (
pl.when(pl.col(amount_name) != 0)
.then(pl.col(discount_name) / pl.col(amount_name))
.otherwise(0)
.alias(discount_in_percent_name)
.cast(pl.Float64)
)
new_columns.extend([expr_price_per_volume_net, expr_discount_percent_2])
# NOTE: The above will override `discountInPercentName` if it was created in BLOCK C.
# If you do NOT want to override, remove or modify this line.
# ============== BLOCK E ==============
# If both margin and amount columns exist, generate ``marginInPercentName``
# using ``marginName / amountName`` when ``amountName`` is non-zero.
if {margin_name, amount_name}.issubset(existing_columns):
expr_margin_in_percent = (
pl.when(pl.col(amount_name) != 0)
.then(pl.col(margin_name) / pl.col(amount_name))
.otherwise(0)
.alias(margin_in_percent_name)
.cast(pl.Float64)
)
new_columns.append(expr_margin_in_percent)
# ============== BLOCK F ==============
# With ``marginName`` and ``netOfDiscountName`` available, compute
# ``marginInPercentOfNetSalesName`` as ``marginName / netOfDiscountName``
# whenever ``netOfDiscountName`` is non-zero.
if {margin_name, net_of_discount_name}.issubset(existing_columns):
expr_margin_in_percent_of_net = (
pl.when(pl.col(net_of_discount_name) != 0)
.then(pl.col(margin_name) / pl.col(net_of_discount_name))
.otherwise(0)
.alias(margin_in_percent_of_net_sales_name)
.cast(pl.Float64)
)
new_columns.append(expr_margin_in_percent_of_net)
# --- 3) Apply the transformations if there are any ---
if new_columns:
df_lazy = df_lazy.with_columns(new_columns)
# --- 4) Return the updated lazy frame (nothing is executed yet) ---
return df_lazy
def get_average_growth_rate(
df, metricToPlot, paramDict, chartDict, valueCols, overlayMetric
) -> pl.LazyFrame:
namingParams = get_naming_params()
metricArrayParams = get_metric_array_params()
priceMetricsArray = metricArrayParams[namingParams["priceMetricsArray"]]
percentMetricsArray = metricArrayParams[namingParams["percentMetricsArray"]]
growthMetricArray = metricArrayParams[namingParams["growthMetricArray"]]
valueMetricsArray = metricArrayParams[namingParams["valueMetricsArray"]]
volumeMetricsArray = metricArrayParams[namingParams["volumeMetricsArray"]]
noSumMetricsArray = metricArrayParams[namingParams["noSumMetricsArray"]]
periodName = namingParams["periodName"]
totalName = namingParams["totalName"]
selectedPeriods = namingParams["selectedPeriods"]
toPlotPeriod = namingParams["toPlotPeriod"]
toPlotPeriod = chartDict[toPlotPeriod]
periodOrder = chartDict[selectedPeriods]
df_lazy = ensure_lazyframe(df)
if metricToPlot in growthMetricArray:
gb = df_lazy.group_by(periodName)
if hasattr(gb, "agg"):
dfSum = gb.agg([pl.col(col).sum().alias(col) for col in valueCols])
else: # compatibility with test wrappers
dfSum = gb[valueCols].sum()
dfSum = dfSum.with_columns(pl.lit(totalName).alias(totalName))
dfSum = get_growth_rate(
dfSum, totalName, periodOrder, paramDict, chartDict, overlayMetric
)
dfSum = dfSum.filter(pl.col(periodName) == toPlotPeriod)
elif metricToPlot in percentMetricsArray + priceMetricsArray:
gb = df_lazy.group_by(periodName)
if hasattr(gb, "agg"):
dfSum = gb.agg([pl.col(col).sum().alias(col) for col in valueCols])
else:
dfSum = gb[valueCols].sum()
dfSum = dfSum.filter(pl.col(periodName) == toPlotPeriod)
dfSum = insert_unit_and_volume_price_column(dfSum)
else:
dfSum = pl.LazyFrame()
return ensure_lazyframe(dfSum)
def get_period_values(dfCopy, chosenDimension, salesMetric):
"""Return period pivoted values as a ``LazyFrame``."""
namingParams = get_naming_params()
periodName = namingParams["periodName"]
df_lazy = ensure_lazyframe(duplicate_dataframe(dfCopy))
valueCols = [salesMetric]
valueCols = check_value_column_exist(df_lazy, valueCols)
if not valueCols:
return pl.LazyFrame()
unique_dims = None
if logger.isEnabledFor(logging.DEBUG):
unique_dims = (
df_lazy.select(pl.col(chosenDimension).unique())
.collect(engine="streaming")
.get_column(chosenDimension)
.to_list()
)
check_collect("APV", "unique_dims", unique_dims)
grouped = df_lazy.group_by([chosenDimension, periodName]).agg(
pl.col(salesMetric).sum().alias(salesMetric)
)
df_lazy = pivot_lazy(
lf=grouped,
index_col=chosenDimension,
pivot_col=periodName,
value_col=salesMetric,
agg_func="first",
)
df_lazy = flatten_cols_polars(df_lazy, "")
df_lazy, newCols = clean_column_labels_after_flatten_df(df_lazy, valueCols)
return df_lazy
def join_period_values(
df: pl.DataFrame | pl.LazyFrame,
dfCopy: pl.DataFrame | pl.LazyFrame,
checkedPeriods: list[str],
chosenDimension: str,
) -> pl.LazyFrame:
"""Join period-based values back to ``dfCopy`` lazily using Polars."""
lf_base = ensure_lazyframe(dfCopy).sort(chosenDimension)
lf_to_join = (
ensure_lazyframe(df).pipe(drop_columns, checkedPeriods).sort(chosenDimension)
)
return lf_base.join(lf_to_join, on=chosenDimension, how="left")
def calculate_sales_growth(
dfCopy: pl.DataFrame | pl.LazyFrame,
chosenDimension: str,
periodOrder: list[str],
salesMetric: str,
growthMetric: str,
) -> pl.LazyFrame:
"""Calculate year-to-year sales growth lazily using Polars."""
lf = ensure_lazyframe(dfCopy)
df_lazy = get_period_values(lf, chosenDimension, salesMetric)
columns, _ = get_schema_and_column_names(df_lazy)
checkedPeriods = []
for p in periodOrder:
if p in columns:
checkedPeriods.append(p)
elif p.lower() in columns:
checkedPeriods.append(p.lower())
elif f"_{p}" in columns:
checkedPeriods.append(f"_{p}")
elif f"_{p.lower()}" in columns:
checkedPeriods.append(f"_{p.lower()}")
else:
checkedPeriods.append(p)
growth_expr = (
(pl.col(checkedPeriods[1]) - pl.col(checkedPeriods[0]))
/ pl.col(checkedPeriods[0])
* 100
).round(0)
df_lazy = (
df_lazy.with_columns(
pl.when(pl.col(checkedPeriods[0]) != 0)
.then(growth_expr)
.otherwise(None)
.alias(growthMetric)
)
.with_columns(
pl.when((pl.col(checkedPeriods[1]) < 0) & (pl.col(checkedPeriods[0]) < 0))
.then(pl.col(growthMetric) * -1)
.otherwise(pl.col(growthMetric))
.alias(growthMetric)
)
.with_columns(
pl.when(pl.col(growthMetric) > 999)
.then(pl.lit(999))
.otherwise(pl.col(growthMetric))
.alias(growthMetric)
)
)
toKeep = checkedPeriods + [growthMetric, chosenDimension]
toDrop = [c for c in columns if c not in toKeep]
df_lazy = drop_columns(df_lazy, toDrop)
df_growth = drop_columns(df_lazy, checkedPeriods)
result = lf.join(df_growth, on=chosenDimension, how="left")
return result
def calculate_price_change(
dfCopy: pl.DataFrame | pl.LazyFrame,
chosenDimension: str,
amountMetric: str,
quantityMetric: str,
priceMetric: str,
periodOrder: list[str],
) -> pl.LazyFrame:
"""Compute price change across periods using Polars lazily."""
lf = ensure_lazyframe(dfCopy)
df_amount = get_period_values(lf, chosenDimension, amountMetric)
df_quantity = get_period_values(lf, chosenDimension, quantityMetric)
df_price = df_amount.join(
df_quantity, on=chosenDimension, how="inner", suffix="_qty"
)
amount_cols, _ = get_schema_and_column_names(df_amount)
period_cols = [c for c in amount_cols if c != chosenDimension]
new_exprs = []
for col in period_cols:
qty_col = f"{col}_qty"
new_exprs.append((pl.col(col) / pl.col(qty_col)).alias(col))
df_price = df_price.with_columns(new_exprs)
df_price = drop_columns(df_price, [f"{c}_qty" for c in period_cols])
columns, _ = get_schema_and_column_names(df_price)
checkedPeriods = []
for p in periodOrder:
if p in columns:
checkedPeriods.append(p)
elif p.lower() in columns:
checkedPeriods.append(p.lower())
elif f"_{p}" in columns:
checkedPeriods.append(f"_{p}")
elif f"_{p.lower()}" in columns:
checkedPeriods.append(f"_{p.lower()}")
else:
checkedPeriods.append(p)
growth_expr = (
(pl.col(checkedPeriods[1]) - pl.col(checkedPeriods[0]))
/ pl.col(checkedPeriods[0])
* 100
).round(0)
df_price = df_price.with_columns(
pl.when(pl.col(checkedPeriods[0]) != 0)
.then(growth_expr)
.otherwise(None)
.alias(priceMetric)
)
df_price = df_price.with_columns(
pl.when((pl.col(checkedPeriods[1]) < 0) & (pl.col(checkedPeriods[0]) < 0))
.then(pl.col(priceMetric) * -1)
.otherwise(pl.col(priceMetric))
.alias(priceMetric)
)
df_price = df_price.with_columns(
pl.when(pl.col(priceMetric) > 999)
.then(pl.lit(999))
.otherwise(pl.col(priceMetric))
.alias(priceMetric)
)
result = join_period_values(df_price, lf, checkedPeriods, chosenDimension)
return result
def get_growth_rate(
dfCopy, chosenDimension, periodOrder, paramDict, chartDict, overlayMetric
):
namingParams = get_naming_params()
metricArrayParams = get_metric_array_params()
priceMetricsArray = metricArrayParams[namingParams["priceMetricsArray"]]
percentMetricsArray = metricArrayParams[namingParams["percentMetricsArray"]]
growthMetricArray = metricArrayParams[namingParams["growthMetricArray"]]
valueMetricsArray = metricArrayParams[namingParams["valueMetricsArray"]]
volumeMetricsArray = metricArrayParams[namingParams["volumeMetricsArray"]]
noSumMetricsArray = metricArrayParams[namingParams["noSumMetricsArray"]]
amountName = namingParams["monetaryLocalCurrencyName"]
volumeName = namingParams["volumeName"]
marginName = namingParams["marginName"]
netOfDiscountName = namingParams["netOfDiscountName"]
unitsName = namingParams["unitsName"]
onePeriodOnly = namingParams["onePeriodOnly"]
notMetConditionValue = namingParams["notMetConditionValue"]
metConditionValue = namingParams["metConditionValue"]
nothingFilteredName = namingParams["nothingFilteredName"]
xAxisMetric = namingParams["xAxisMetric"]
yAxisMetric = namingParams["yAxisMetric"]
singleMetric = namingParams["singleMetric"]
metricsToPlot = namingParams["metricsToPlot"]
volumePriceChangeName = namingParams["volumePriceChangeName"]
netVolumePriceChangeName = namingParams["netVolumePriceChangeName"]
unitsPriceChangeName = namingParams["unitsPriceChangeName"]
netUnitsPriceChangeName = namingParams["netUnitsPriceChangeName"]
salesGrowthName = namingParams["salesGrowthName"]
netSalesGrowthName = namingParams["netSalesGrowthName"]
unitsGrowthName = namingParams["unitsGrowthName"]
volumeGrowthName = namingParams["volumeGrowthName"]
marginGrowthName = namingParams["marginGrowthName"]
chosenChart = namingParams["chosenChart"]
calculateGrowth = notMetConditionValue
if xAxisMetric in chartDict and yAxisMetric in chartDict:
xAxisMetric = chartDict[xAxisMetric]
yAxisMetric = chartDict[yAxisMetric]
calculateGrowth = metConditionValue
elif singleMetric in chartDict:
xAxisMetric = chartDict[singleMetric]
yAxisMetric = chartDict[singleMetric]
calculateGrowth = metConditionValue
elif metricsToPlot in chartDict and len(chartDict[metricsToPlot]) > 0:
xAxisMetric = chartDict[metricsToPlot][overlayMetric]
yAxisMetric = chartDict[metricsToPlot][overlayMetric]
calculateGrowth = metConditionValue
if calculateGrowth:
if onePeriodOnly in paramDict and paramDict[onePeriodOnly]:
pass
elif chosenDimension in [nothingFilteredName, None]:
pass
elif get_row_count(dfCopy) == 0:
pass
elif xAxisMetric in growthMetricArray or yAxisMetric in growthMetricArray:
if xAxisMetric == salesGrowthName or yAxisMetric == salesGrowthName:
dfCopy = calculate_sales_growth(
dfCopy, chosenDimension, periodOrder, amountName, salesGrowthName
)
if xAxisMetric == netSalesGrowthName or yAxisMetric == netSalesGrowthName:
dfCopy = calculate_sales_growth(
dfCopy,
chosenDimension,
periodOrder,
netOfDiscountName,
netSalesGrowthName,
)
if xAxisMetric == unitsGrowthName or yAxisMetric == unitsGrowthName:
dfCopy = calculate_sales_growth(
dfCopy, chosenDimension, periodOrder, unitsName, unitsGrowthName
)
if xAxisMetric == volumeGrowthName or yAxisMetric == volumeGrowthName:
dfCopy = calculate_sales_growth(
dfCopy, chosenDimension, periodOrder, volumeName, volumeGrowthName
)
if xAxisMetric == marginGrowthName or yAxisMetric == marginGrowthName:
dfCopy = calculate_sales_growth(
dfCopy, chosenDimension, periodOrder, marginName, marginGrowthName
)
if (
xAxisMetric == unitsPriceChangeName
or yAxisMetric == unitsPriceChangeName
):
dfCopy = calculate_price_change(
dfCopy,
chosenDimension,
amountName,
unitsName,
unitsPriceChangeName,
periodOrder,
)
if (
xAxisMetric == netUnitsPriceChangeName
or yAxisMetric == netUnitsPriceChangeName
):
dfCopy = calculate_price_change(
dfCopy,
chosenDimension,
netOfDiscountName,
unitsName,
netUnitsPriceChangeName,
periodOrder,
)
if (
xAxisMetric == volumePriceChangeName
or yAxisMetric == volumePriceChangeName
):
dfCopy = calculate_price_change(
dfCopy,
chosenDimension,
amountName,
volumeName,
volumePriceChangeName,
periodOrder,
)
if (
xAxisMetric == netVolumePriceChangeName
or yAxisMetric == netVolumePriceChangeName
):
dfCopy = calculate_price_change(
dfCopy,
chosenDimension,
netOfDiscountName,
volumeName,
netVolumePriceChangeName,
periodOrder,
)
return dfCopy
def multiply_percent_metrics_by_hundred(df):
namingParams = get_naming_params()
metricArrayParams = get_metric_array_params()
priceMetricsArray = metricArrayParams[namingParams["priceMetricsArray"]]
percentMetricsArray = metricArrayParams[namingParams["percentMetricsArray"]]
growthMetricArray = metricArrayParams[namingParams["growthMetricArray"]]
valueMetricsArray = metricArrayParams[namingParams["valueMetricsArray"]]
volumeMetricsArray = metricArrayParams[namingParams["volumeMetricsArray"]]
noSumMetricsArray = metricArrayParams[namingParams["noSumMetricsArray"]]
columns, schema = get_schema_and_column_names(df)
# Use Polars expressions to avoid pandas-style direct assignment
exprs = []
for metric in percentMetricsArray:
if metric in columns:
exprs.append(((pl.col(metric) * 100).round(1)).alias(metric))
if exprs:
return df.with_columns(exprs)
return df
def adjust_percentages_dynamic(
df: pl.DataFrame | pl.LazyFrame,
) -> pl.DataFrame | pl.LazyFrame:
"""
For each row:
1) Keep the leftmost_col unchanged (often a text label).
2) For each numeric column (excluding leftmost_col, total_col),
compute an integer %: int(round((col_i / row[total_col]) * 100)).
3) Sum these integer %'s; leftover = 100 - that sum.
4) Add leftover to the single largest column in that row (may be + or -).
5) Overwrite 'total_col' with 100.
Returns a new Polars DataFrame with integer percentages that sum to 100.
"""
namingParams = get_naming_params()
leftmost_col = namingParams["periodName"]
total_col = namingParams["valueName"]
numeric_types = {
pl.Int8,
pl.Int16,
pl.Int32,
pl.Int64,
pl.UInt8,
pl.UInt16,
pl.UInt32,
pl.UInt64,
pl.Float32,
pl.Float64,
}
skip_cols = {leftmost_col, total_col}
schema = get_schema_and_column_names(df)[1]
numeric_cols = [
name
for name, dtype in schema.items()
if name not in skip_cols and dtype in numeric_types
]
if not numeric_cols:
return df.with_columns(pl.lit(100).alias(total_col))
frame = df
percent_exprs = [
pl.when(pl.col(total_col) == 0)
.then(0)
.otherwise((pl.col(col) / pl.col(total_col) * 100).round(0))
.cast(pl.Int64)
.alias(col)
for col in numeric_cols
]
frame = frame.with_columns(percent_exprs)
sum_expr = pl.fold(
pl.lit(0), lambda acc, e: acc + e, [pl.col(c) for c in numeric_cols]
).alias("__sum__")
frame = frame.with_columns(sum_expr)
diff_expr = (
pl.when(pl.col(total_col) == 0)
.then(0)
.otherwise(100 - pl.col("__sum__"))
.alias("__diff__")
)
frame = frame.with_columns(diff_expr)
frame = frame.with_columns(
pl.concat_list([pl.col(c) for c in numeric_cols]).alias("__list__")
)
frame = frame.with_columns(pl.col("__list__").list.arg_max().alias("__argmax__"))
updates = [
pl.when(pl.col("__argmax__") == idx)
.then(pl.col(col) + pl.col("__diff__"))
.otherwise(pl.col(col))
.alias(col)
for idx, col in enumerate(numeric_cols)
]
frame = frame.with_columns(updates)
frame = frame.with_columns(pl.lit(100).alias(total_col))
return frame.drop(["__sum__", "__diff__", "__list__", "__argmax__"])
def replace_NA_with_blanc(lf, column):
lf = lf.with_columns(
pl.col(column)
.str.replace(r"^(N/A|NaN|)$", None) # Match and replace
.alias(column)
)
return lf
def lookup_dates_polars(lf: pl.LazyFrame, col: str) -> pl.LazyFrame:
"""Parse unique date strings once and cast the column."""
try:
unique_dates = (
lf.select(pl.col(col).drop_nulls().unique()).collect()[col].to_list()
)
check_collect("TAA", "unique_dates", unique_dates)
lf = replace_NA_with_blanc(lf, col)
parse_expr = (
pl.col(col)
.str.to_datetime(strict=False)
.dt.date()
.fill_null(pl.col(col).str.strptime(pl.Date, "%Y-%m-%d", strict=False))
.fill_null(pl.col(col).str.strptime(pl.Date, "%Y/%m/%d", strict=False))
.fill_null(pl.col(col).str.strptime(pl.Date, "%d-%m-%Y", strict=False))
.fill_null(pl.col(col).str.strptime(pl.Date, "%b %d %Y", strict=False))
)
lf = lf.with_columns(parse_expr.alias(col))
except Exception as e: # pragma: no cover - unforeseen date formats
logging.exception(e)
logger.exception("Parsing failed for column '%s'", col)
raise ValueError(f"Failed to parse dates in column '{col}'") from e
return lf
def check_no_duplicates(array):
newArray = []
for element in array:
if element not in newArray:
newArray.append(element)
return newArray
def check_if_more_than_one_in_common(findInArray, findFromArray):
"""
if there are more than one hierarchical node of a same hierarchy
we only want to keep the most detailed node
"""
both = set(findInArray).intersection(findFromArray)
indices_findInArray = [findInArray.index(x) for x in both]
indices_findInArray.sort()
if len(indices_findInArray) > 0:
indices_findInArray.pop()
count = 0
for element in indices_findInArray:
element = element - count
findInArray.pop(element)
count = count + 1
return findInArray
def insert_missing_parents(findInArray, findFromArray):
"""
in this alternative method we complete the array with all their parents
"""
findFromArray = list(findFromArray.keys())
both = set(findInArray).intersection(findFromArray)
indices_findFromArray = [findFromArray.index(x) for x in both]
indices_findFromArray.sort()
if len(indices_findFromArray) > 0:
maxHierarchyIndex = indices_findFromArray[-1]
fullHierarchicalIndexes = findFromArray[: maxHierarchyIndex + 1]
findInArray = findInArray + fullHierarchicalIndexes
findInArray = list(set(findInArray))
return findInArray
def add_bold(string):
string = "**" + string + "**"
return string
def add_bold_and_carrier_return(string):
string = "**" + string + "**" + "\n"
return string
def to_excel(df: pl.DataFrame | pl.LazyFrame) -> bytes:
"""Return an Excel representation of ``df`` using Polars.
Complex objects like ``dict`` or ``list`` are serialized to JSON strings so
cells remain readable when opened in Excel.
"""
if isinstance(df, pl.LazyFrame):
df_pl = df.collect()
else:
df_pl = df
def _convert(val: Any) -> str:
if isinstance(val, (dict, list)):
return json.dumps(val, ensure_ascii=False)
return str(val)
exprs = []
df_pl_cols, _ = get_schema_and_column_names(df_pl)
for col, dtype in zip(df_pl_cols, df_pl.dtypes):
if dtype in (pl.Object, pl.Struct) or isinstance(dtype, pl.List):
exprs.append(
pl.col(col).map_elements(_convert, return_dtype=pl.Utf8).alias(col)
)
else:
exprs.append(pl.col(col))
df_pl = df_pl.select(exprs)
output = BytesIO()
write_polars_excel(df_pl, output)
return output.getvalue()
def get_table_download_link(data, name, fileFormat):
"""Return a download link for ``data``.
``data`` may be a :class:`polars.DataFrame` or :class:`polars.LazyFrame`.
"""
if fileFormat == "Excel":
val = to_excel(data)
b64 = base64.b64encode(val)
return (
f'<a href="data:application/octet-stream;base64,{b64.decode()}" '
f'download="{name}.xlsx">Download "{name}" Excel file.</a>'
)
elif fileFormat == "csv":
df_pl = data.collect() if isinstance(data, pl.LazyFrame) else data
val = df_pl.write_csv()
b64 = base64.b64encode(val.encode()).decode()
return (
f'<a href="data:file/csv;base64,{b64}" '
f'download="{name}.csv">Download "{name}" csv file</a>'
)
elif fileFormat == "profile":
with open(data, "rb") as f:
val = f.read()
b64 = base64.b64encode(val).decode()
return (
f'<a href="data:file/csv;base64,{b64}" download="'
+ name
+ '.html">Download "'
+ name
+ '" html file</a>'
) # decode b'abc' => abc
def get_table_number_format(df, colArray):
"""
the smaller the number the more decimal positions we want
"""
namingParams = get_naming_params()
columns, schema = get_schema_and_column_names(df)
numberFormatDict = {}
numberFormatDict["{:,.0f}"] = []
numberFormatDict["{:,.1f}"] = []
numberFormatDict["{:,.2f}"] = []
numberFormatDict["{:,.3f}"] = []
for column in colArray:
if column in columns:
# Sum of first 5 rows for this column (pandas-like head().sum())
columnSum = (
ensure_lazyframe(df)
.select(pl.col(column))
.head(5)
.select(pl.col(column).sum())
.collect()
.item()
)
absoluteSum = math.fabs(columnSum)
if absoluteSum < 0.1:
numberFormatDict["{:,.2f}"].append(column)
elif absoluteSum < 1:
numberFormatDict["{:,.2f}"].append(column)
elif absoluteSum < 10:
numberFormatDict["{:,.1f}"].append(column)
elif absoluteSum < 100:
numberFormatDict["{:,.1f}"].append(column)
else:
numberFormatDict["{:,.0f}"].append(column)
return numberFormatDict
def get_row_data_from_original_df(dfDict, dfCopy, chartDict, indexCols):
"""
based on the plotting choice of the user we want to return a dataframe
with the elements corresponding to that report row
"""
namingParams = get_naming_params()
dfMainReportName = namingParams["dfMainReportName"]
rowToPlot = namingParams["rowToPlotName"]
nanFillValue = namingParams["nanFillValue"]
workColumn = namingParams["workColumn"]
varianceType = namingParams["varianceTypeName"]
plotOriginalData = namingParams["plotOriginalData"]
notMetConditionValue = namingParams["notMetConditionValue"]
dfDictListDrilledName = namingParams["dfDictListDrilledName"]
drillDownDatasetNumber = namingParams["drillDownDatasetNumber"]
mainReportRunName = namingParams["mainReportRunName"]
chosenRow = chartDict[rowToPlot] - 1
if (
drillDownDatasetNumber in chartDict
and chartDict[drillDownDatasetNumber] != mainReportRunName
):
dfListCopy = dfDict[dfDictListDrilledName][chartDict[drillDownDatasetNumber]]
else:
dfListCopy = dfDict[dfMainReportName]
dfList = duplicate_dataframe(dfListCopy)
filterDict: dict[str, list] = {}
if is_valid_lazyframe(dfList):
columns, _ = get_schema_and_column_names(dfList)
filterCols = indexCols + [varianceType]
foundFilterCols = [c for c in filterCols if c in columns]
lf_list = (
ensure_lazyframe(dfList)
.select(foundFilterCols)
.drop_nulls()
.limit(chosenRow + 1)
)
df_temp = lf_list.collect()
df_temp_cols, _ = get_schema_and_column_names(df_temp)
filterDict = {col: df_temp[col].to_list() for col in df_temp_cols}
if chosenRow + 1 > df_temp.height:
chosenRow = df_temp.height - 1
chartDict[rowToPlot] = chosenRow + 1
rowVarianceType = filterDict.get(varianceType, [None] * (chosenRow + 1))[
chartDict[rowToPlot] - 1
]
df = ensure_lazyframe(duplicate_dataframe(dfCopy))
df = df.with_columns(pl.lit(rowVarianceType).alias(varianceType))
for row in range(chosenRow + 1):
exprs = []
for column, values in filterDict.items():
value = values[row]
if value != nanFillValue:
exprs.append(pl.col(column).is_in([value]))
if exprs:
condition = exprs[0]
for e in exprs[1:]:
condition &= e
if row == chosenRow:
df = df.filter(condition)
elif (
plotOriginalData in chartDict
and chartDict[plotOriginalData] == notMetConditionValue
):
df = df.filter(~condition)
return df, len(filterDict), chartDict
def check_if_other_in_x_dimension(df, chartDict):
namingParams = get_naming_params()
aggregateOtherItemsName = namingParams["aggregateOtherItemsName"]
xAxisDimension = namingParams["xAxisDimension"]
isOtherRank = False
itemArray = (
ensure_lazyframe(df)
.select(pl.col(chartDict[xAxisDimension]).cast(pl.Utf8).unique())
.collect()
.get_column(chartDict[xAxisDimension])
.to_list()
)
for element in itemArray:
if aggregateOtherItemsName in element:
isOtherRank = True
return isOtherRank
def check_if_other_in_index(df):
namingParams = get_naming_params()
aggregateOtherItemsName = namingParams["aggregateOtherItemsName"]
# Polars has no implicit index; treat the first column as the "index-like" label.
columns, _ = get_schema_and_column_names(df)
if not columns:
return False
first_col = columns[0]
try:
items = (
ensure_lazyframe(df)
.select(pl.col(first_col).cast(pl.Utf8).unique())
.collect()
.get_column(first_col)
.to_list()
)
except Exception as e:
logging.exception(e)
# Fallback for eager DataFrame
if isinstance(df, pl.DataFrame):
items = df.get_column(first_col).cast(pl.Utf8).unique().to_list()
else:
items = []
for element in items:
if element is not None and aggregateOtherItemsName in element:
return True
return False
def check_if_other_in_columns(df):
namingParams = get_naming_params()
aggregateOtherItemsName = namingParams["aggregateOtherItemsName"]
isOtherRank = False
columns, schema = get_schema_and_column_names(df)
for element in columns:
if aggregateOtherItemsName in element:
isOtherRank = True
return isOtherRank
def check_if_other_in_rows(df, column):
namingParams = get_naming_params()
aggregateOtherItemsName = namingParams["aggregateOtherItemsName"]
xAxisDimension = namingParams["xAxisDimension"]
isOtherRank = False
columns, schema = get_schema_and_column_names(df)
if column in columns:
itemArray = (
ensure_lazyframe(df)
.select(pl.col(column).cast(pl.Utf8).unique())
.collect()
.get_column(column)
.to_list()
)
for element in itemArray:
if aggregateOtherItemsName in element:
isOtherRank = True
return isOtherRank
SHA-256: 3381fcbd8f2dfb14462605fec885fba4885d9ebcf456e8722df99af21ee4d96d