import copy
import logging
from functools import reduce
from itertools import combinations

import polars as pl

from modules.data.common_data_utils import (
    complete_combination_with_all_parents,
    delete_hierarchical_parents,
)
from modules.layout.layout_data import collect_lazyframe
from modules.layout.set_up_widgets import submit_variance_charts
from modules.utilities.config import (
    get_config_params,
    get_naming_params,
    get_run_params,
    get_variance_aggregation_params,
)
from modules.utilities.helpers import (
    add_status_message_to_paramDict,
    check_if_duplicates_in_all_columns,
    drop_columns,
    duplicate_dataframe,
    get_data_sample,
    group_by_df_on_index_cols,
    measure_time,
    move_to_end,
    round_value_columns_to_dec,
    take_filtered_value_out_of_option_list,
    unique,
)
from modules.utilities.ui_notifier import ui
from modules.utilities.utils import (
    ensure_polars_df,
    get_schema_and_column_names,
    is_valid_lazyframe,
)

try:
    from modules.utilities.utils import ensure_lazyframe
except ImportError:  # pragma: no cover

    def ensure_lazyframe(obj: pl.DataFrame | pl.LazyFrame) -> pl.LazyFrame:
        if isinstance(obj, pl.LazyFrame):
            return obj
        if isinstance(obj, pl.DataFrame):
            return obj.lazy()
        return pl.DataFrame(obj).lazy()


from modules.variance.variance_formulas import (
    calculate_margin_mix_variance,
    calculate_sales_mix_variance,
    calculate_variance_in_percent,
)
from modules.variance.variance_utils import (
    add_random_key,
    change_nan_to_all,
    divide_back_if_multiplied,
    filter_by_number_of_nodes,
    insert_dates,
    make_divideArray,
    recalculate_price,
)


def build_variance_calculation_array(aggregationsToPlot, chartDict):
    namingParams = get_naming_params()
    totalVarianceAggregation = namingParams["totalVarianceAggregation"]
    priceAndUnitsAggregation = namingParams["priceAndUnitsAggregation"]
    priceAndVolumeAggregation = namingParams["priceAndVolumeAggregation"]
    mixAndUnitsAggregation = namingParams["mixAndUnitsAggregation"]
    mixAndVolumeAggregation = namingParams["mixAndVolumeAggregation"]
    costsUnitsAggregation = namingParams["costsUnitsAggregation"]
    costsVolumeAggregation = namingParams["costsVolumeAggregation"]
    costsUnitsMixAggregation = namingParams["costsUnitsMixAggregation"]
    costsVolumeMixAggregation = namingParams["costsVolumeMixAggregation"]
    marginUnitsRateAggregation = namingParams["marginUnitsRateAggregation"]
    marginVolumeRateAggregation = namingParams["marginVolumeRateAggregation"]
    varianceArray = []
    aggregationsToPlotDict = {
        totalVarianceAggregation: ["Price & units & mix"],
        priceAndUnitsAggregation: ["Price", "Units & mix"],
        priceAndVolumeAggregation: ["Price", "Volume & mix"],
        mixAndUnitsAggregation: ["Price", "Units", "Mix"],
        mixAndVolumeAggregation: ["Price", "Volume", "Mix"],
        costsUnitsAggregation: [
            "Price on margin",
            "Units & mix on margin",
            "Cost",
            "Indirect Costs",
            "Balance",
        ],
        costsVolumeAggregation: [
            "Price on margin",
            "Volume & mix on margin",
            "Cost",
            "Indirect Costs",
            "Balance",
        ],
        costsUnitsMixAggregation: [
            "Price on margin",
            "Units on margin",
            "Mix on margin",
            "Cost",
            "Indirect Costs",
            "Balance",
        ],
        costsVolumeMixAggregation: [
            "Price on margin",
            "Volume on margin",
            "Mix on margin",
            "Cost",
            "Indirect Costs",
            "Balance",
        ],
        marginUnitsRateAggregation: [
            "Price on margin",
            "Units & mix on margin",
            "Margin rate",
            "Indirect Costs",
            "Balance",
        ],
        marginVolumeRateAggregation: [
            "Price on margin",
            "Volume & mix on margin",
            "Margin rate",
            "Indirect Costs",
            "Balance",
        ],
    }
    for element in aggregationsToPlot:
        if element in aggregationsToPlotDict:
            for item in aggregationsToPlotDict[element]:
                if item not in varianceArray:
                    varianceArray.append(item)
    varianceArray = move_to_end(varianceArray, "Balance")
    varianceArray = insert_dates(chartDict, varianceArray)
    return varianceArray


def select_variance_aggregations_to_plot(chartDict):
    namingParams = get_naming_params()
    varianceAggregationParams = get_variance_aggregation_params()
    cogsAggregationArray = varianceAggregationParams[
        namingParams["cogsAggregationArray"]
    ]
    salesAggregationArray = varianceAggregationParams[
        namingParams["salesAggregationArray"]
    ]
    discountsAggregationArray = varianceAggregationParams[
        namingParams["discountsAggregationArray"]
    ]
    varianceAggregationOptionsArrayKey = namingParams["varianceAggregationOptionsArray"]
    varianceAggregationKey = namingParams["varianceAggregation"]
    marginVarianceAggregation = namingParams["marginVarianceAggregation"]
    marginUnitsRateAggregation = namingParams["marginUnitsRateAggregation"]
    marginVolumeRateAggregation = namingParams["marginVolumeRateAggregation"]
    costsUnitsAggregation = namingParams["costsUnitsAggregation"]
    costsVolumeAggregation = namingParams["costsVolumeAggregation"]
    costsUnitsMixAggregation = namingParams["costsUnitsMixAggregation"]
    costsVolumeMixAggregation = namingParams["costsVolumeMixAggregation"]
    discountsUnitsCogsAggregation = namingParams["discountsUnitsCogsAggregation"]
    discountsVolumeCogsAggregation = namingParams["discountsVolumeCogsAggregation"]
    netOfDiscountAggregation = namingParams["netOfDiscountAggregation"]
    discountsAndUnitsAggregation = namingParams["discountsAndUnitsAggregation"]
    discountsAndVolumeAggregation = namingParams["discountsAndVolumeAggregation"]
    discountsAggregation = namingParams["discountsAggregation"]
    cogsAggregation = namingParams["cogsAggregation"]
    totalVarianceAggregation = namingParams["totalVarianceAggregation"]
    priceAndUnitsAggregation = namingParams["priceAndUnitsAggregation"]
    priceAndVolumeAggregation = namingParams["priceAndVolumeAggregation"]
    unitPriceOnSalesAggregation = namingParams["unitPriceOnSalesAggregation"]
    volumePriceOnSalesAggregation = namingParams["volumePriceOnSalesAggregation"]
    unitsOnSalesAggregation = namingParams["unitsOnSalesAggregation"]
    volumeOnSalesAggregation = namingParams["volumeOnSalesAggregation"]
    unitPriceOnMarginAggregation = namingParams["unitPriceOnMarginAggregation"]
    volumePriceOnMarginAggregation = namingParams["volumePriceOnMarginAggregation"]
    unitsOnMarginAggregation = namingParams["unitsOnMarginAggregation"]
    volumeOnMarginAggregation = namingParams["volumeOnMarginAggregation"]
    newAndLostUnitsAggregation = namingParams["newAndLostUnitsAggregation"]
    newAndLostVolumeAggregation = namingParams["newAndLostVolumeAggregation"]
    newAndLostUnitsMixAggregation = namingParams["newAndLostUnitsMixAggregation"]
    newAndLostVolumeMixAggregation = namingParams["newAndLostVolumeMixAggregation"]
    mixAndUnitsAggregation = namingParams["mixAndUnitsAggregation"]
    mixAndVolumeAggregation = namingParams["mixAndVolumeAggregation"]
    metConditionValue = namingParams["metConditionValue"]
    notMetConditionValue = namingParams["notMetConditionValue"]
    varianceDifferentCalculations = namingParams["varianceDifferentCalculations"]
    salesAggregationsToPlot = [
        totalVarianceAggregation,
        priceAndUnitsAggregation,
        priceAndVolumeAggregation,
        mixAndUnitsAggregation,
        mixAndVolumeAggregation,
    ]
    marginAggregationsToPlot = [
        marginUnitsRateAggregation,
        costsUnitsAggregation,
        costsVolumeAggregation,
        costsUnitsMixAggregation,
        costsVolumeMixAggregation,
    ]
    aggregationsToPlot = []
    cogsAggregation = notMetConditionValue
    chartDict[varianceDifferentCalculations] = notMetConditionValue
    if varianceAggregationOptionsArrayKey in chartDict:
        varianceAggregationOptionsArray = chartDict[varianceAggregationOptionsArrayKey]
    if varianceAggregationKey in chartDict:
        varianceAggregation = chartDict[varianceAggregationKey]
        if varianceAggregation in salesAggregationArray:
            for element in varianceAggregationOptionsArray:
                if element in salesAggregationsToPlot:
                    aggregationsToPlot.append(element)
        elif varianceAggregation in cogsAggregationArray:
            cogsAggregation = metConditionValue
            for element in varianceAggregationOptionsArray:
                if element in marginAggregationsToPlot:
                    aggregationsToPlot.append(element)
    if len(aggregationsToPlot) > 1:
        chartDict[varianceDifferentCalculations] = metConditionValue
        chartDict[varianceAggregationOptionsArrayKey] = aggregationsToPlot
        if cogsAggregation:
            if marginUnitsRateAggregation in aggregationsToPlot:
                chartDict[varianceAggregationKey] = marginUnitsRateAggregation
            elif marginVolumeRateAggregation in aggregationsToPlot:
                chartDict[varianceAggregationKey] = marginVolumeRateAggregation
    return aggregationsToPlot, chartDict


def prepare_config_for_variance_plot():
    namingParams = get_naming_params()
    configParams = get_config_params()
    waterfallChart = namingParams["verticalWaterfallChart"]
    configPlotlyDict = configParams["configPlotlyDict"]
    configPlotlyDict = configPlotlyDict[waterfallChart]
    return configPlotlyDict


def set_up_different_variance_calculations_chart(columnArray, paramDict, chartDict):
    bridgeSubmit, chartDict = submit_variance_charts(columnArray, paramDict, chartDict)
    aggregationsToPlot, chartDict = select_variance_aggregations_to_plot(chartDict)
    varianceArray = build_variance_calculation_array(aggregationsToPlot, chartDict)
    configPlotlyDict = prepare_config_for_variance_plot()
    return bridgeSubmit, chartDict, aggregationsToPlot, varianceArray, configPlotlyDict


def find_unique_node_chains(dfCopy, indexCols, paramDict):
    """Return unique node chains keeping the entry with the most nodes."""
    namingParams = get_naming_params()
    numberOfNodes = namingParams["numberOfNodes"]
    uniqueValuesInCombination = namingParams["uniqueValuesInCombination"]
    dataPreparation = namingParams["dataPreparationName"]
    randomKey = namingParams["randomKey"]
    dfNodes = duplicate_dataframe(dfCopy)
    dfNodes, paramDict = group_by_df_on_index_cols(
        dfNodes,
        [randomKey],
        [numberOfNodes, uniqueValuesInCombination],
        "max",
        paramDict,
        False,
    )

    joincols = [randomKey, numberOfNodes, uniqueValuesInCombination]
    # We don't need to set or reset indexes in Polars.
    # Just select the necessary columns from dfCopy:
    # Make sure joincols are included so that the join works correctly.
    dfCopy = dfCopy.select(joincols + indexCols)
    lfCopy = dfCopy.lazy()

    # Perform the join on the specified join columns
    # Note: By default, Polars uses a hash join which doesn't require sorting.
    dfNodes = dfNodes.join(lfCopy, on=joincols, how="left")
    lfNodes = dfNodes.lazy()

    group_byCols = [randomKey, numberOfNodes, uniqueValuesInCombination]

    # First: forward fill within groups
    # Sort by group_byCols so that forward fill happens in a defined order
    dfNodes = lfNodes.sort(group_byCols).with_columns(
        [pl.col(c).fill_null(strategy="forward").over(group_byCols) for c in indexCols]
    )

    # Second: backward fill within the same groups
    # We do another sort to ensure the order is correct for backward filling
    dfNodes = lfNodes.sort(group_byCols).with_columns(
        [pl.col(c).fill_null(strategy="backward").over(group_byCols) for c in indexCols]
    )

    # Drop duplicates
    dfNodes = dfNodes.unique(maintain_order=True)

    # Finally, sort by randomKey (no set_index, just sort)
    dfNodes = dfNodes.sort(randomKey)

    return dfNodes, paramDict


def get_metric_values_for_each_random_key(dfCopy, valueCols, paramDict):
    """Aggregate metric columns for each unique random key."""
    namingParams = get_naming_params()
    randomKey = namingParams["randomKey"]
    dfMetrics = duplicate_dataframe(dfCopy)
    metricCols = copy.deepcopy(valueCols)
    metricCols = take_filtered_value_out_of_option_list(metricCols, randomKey)
    dfMetrics, paramDict = group_by_df_on_index_cols(
        dfMetrics, metricCols, [randomKey], "max", paramDict, False
    )
    dfMetrics = dfMetrics.sort(randomKey)
    if isinstance(dfCopy, pl.LazyFrame) and isinstance(dfMetrics, pl.DataFrame):
        dfMetrics = dfMetrics.lazy()
    return dfMetrics, metricCols, paramDict


def delete_duplicate_nodes(dfCopy, indexCols, valueCols, paramDict):
    """Remove duplicate chains and merge the corresponding metrics."""
    namingParams = get_naming_params()
    numberOfNodes = namingParams["numberOfNodes"]
    uniqueValuesInCombination = namingParams["uniqueValuesInCombination"]
    deletedDuplicateNodes = namingParams["deletedDuplicateNodesName"]
    dataPreparation = namingParams["dataPreparationName"]
    findUniqueNodeChains = namingParams["findUniqueNodeChainsName"]
    getMetricValues = namingParams["getMetricValuesName"]
    randomKey = namingParams["randomKey"]
    dfMetrics, metricCols, paramDict = get_metric_values_for_each_random_key(
        dfCopy, valueCols, paramDict
    )
    measure_time(dataPreparation, getMetricValues, False)
    dfNodes, paramDict = find_unique_node_chains(dfCopy, indexCols, paramDict)
    measure_time(dataPreparation, findUniqueNodeChains, False)
    dfResult = dfNodes.join(dfMetrics, on=randomKey, how="left")
    # Fill nulls in metric columns using Polars expressions
    dfResult = dfResult.with_columns(
        [pl.col(c).fill_null(0).alias(c) for c in metricCols]
    )
    # No index semantics in Polars; reset_index is a no-op here
    paramDict = check_if_duplicates_in_all_columns(
        dfResult, "dfResult in delete duplicate rows", paramDict
    )
    dfResult = dfResult.unique(subset=indexCols, keep="first", maintain_order=True)
    measure_time(dataPreparation, deletedDuplicateNodes, False)
    return dfResult, paramDict


def process_index_combinations(dfCopy, resultArray, indexCols, valueCols, paramDict):
    """
    we want to filter out the rows that represent marginal part of total
    """
    configParams = get_config_params()
    namingParams = get_naming_params()
    runParams = get_run_params()
    includeBaseDataframe = runParams["includeBaseDataframe"]
    filterOutPercent = namingParams["filterOutPercent"]
    dataPreparation = namingParams["dataPreparationName"]
    appliedMultiIndexName = namingParams["appliedMultiIndexName"]
    processIndexCombinationName = namingParams["processIndexCombinationName"]
    metConditionValue = namingParams["metConditionValue"]
    notMetConditionValue = namingParams["notMetConditionValue"]
    randomKey = namingParams["randomKey"]
    aggregationValueCols = [column for column in valueCols if column != randomKey]
    dfCopy = drop_columns(dfCopy, [randomKey])
    use_lazy = isinstance(dfCopy, pl.LazyFrame)
    if use_lazy:
        frameArray = []
    elif includeBaseDataframe:
        frameArray = [dfCopy]
    else:
        frameArray = [dfCopy[0:0]]
    for indexCol in resultArray:
        df = duplicate_dataframe(dfCopy)
        if is_valid_lazyframe(df):
            df, paramDict = group_by_df_on_index_cols(
                df, indexCol, aggregationValueCols, "sum", paramDict, False
            )
            if is_valid_lazyframe(df):
                if use_lazy:
                    df = df.lazy() if isinstance(df, pl.DataFrame) else df
                frameArray.append(df)
    measure_time(dataPreparation, processIndexCombinationName, False)
    if len(frameArray) > 0:
        contains_lazy = any(isinstance(f, pl.LazyFrame) for f in frameArray)
        if contains_lazy:
            frameArray = [
                f.lazy() if isinstance(f, pl.DataFrame) else f for f in frameArray
            ]
            dfResult = pl.concat(frameArray, how="diagonal", parallel=False)
        else:
            dfResult = pl.concat(frameArray, how="diagonal")
        paramDict = check_if_duplicates_in_all_columns(
            dfResult, "Process index combinations", paramDict
        )
        columns, _ = get_schema_and_column_names(dfResult)
        stableSortCols = [
            column
            for column in [*indexCols, *aggregationValueCols]
            if column in columns
        ]
        if stableSortCols:
            dfResult = dfResult.sort(stableSortCols, nulls_last=True)
        dfResult = dfResult.unique(subset=indexCols, maintain_order=True)
        # A row key identifies one completed node combination. Summing source
        # row indexes during aggregation creates collisions between parent and
        # child nodes and can attach the wrong metrics to a dimension label.
        keyValueCols = list(aggregationValueCols)
        dfResult, keyValueCols = add_random_key(dfResult, keyValueCols, False)
        if randomKey not in valueCols:
            valueCols.append(randomKey)
        statusMessage = "collect but not cached"
        paramDict = add_status_message_to_paramDict(paramDict, statusMessage, 0)
        if not use_lazy and isinstance(dfResult, pl.LazyFrame):
            dfResult = dfResult.collect()
        dfResult, paramDict = recalculate_price(dfResult, paramDict)
        measure_time(dataPreparation, appliedMultiIndexName, False)
    else:
        dfResult = pl.DataFrame() if not use_lazy else pl.DataFrame().lazy()
    return dfResult, paramDict


def get_index_combinations(indexCols, nodeDict, paramDict):
    """
    returns all combinations of lenghts up to a given number
    """
    namingParams = get_naming_params()
    maxIndexArrayLength = paramDict[namingParams["maxIndexArrayLength"]]
    arrayLength = len(indexCols)
    if arrayLength > maxIndexArrayLength:
        arrayLength = maxIndexArrayLength
    resultArray = []
    for x in range(1, arrayLength + 1):
        resultArray = get_combinations_of_given_length(
            indexCols, x, nodeDict, resultArray
        )
    resultArray.sort()
    return resultArray, paramDict


def get_combinations_of_given_length(indexCols, length, indexDict, resultArray):
    """
    return the combinations of indexes of a given length
    having taken out the non informative hierarchical nodes from each array
    we filter out array with the same element repeated
    """
    runParams = get_run_params()
    deleteHierarchicalParents = runParams["deleteHierarchicalParents"]
    comb = combinations(indexCols, length)
    for i in list(comb):
        array = list(i)
        if deleteHierarchicalParents:
            array = delete_hierarchical_parents(array, indexDict)
        else:
            array = complete_combination_with_all_parents(array, indexDict)
        if array not in resultArray:
            if len(array) > 0:
                if len(array) == 1 or (array.count(array[0]) != len(array)):
                    noDuplicates = unique(array)
                    if noDuplicates not in resultArray:
                        resultArray.append(noDuplicates)
    return resultArray


def process_variance_calculation(dfDict, paramDict, chartDict, indexCols):
    namingParams = get_naming_params()
    varianceAmount = namingParams["varianceAmountName"]
    dfBaseKey = namingParams["dfBaseName"]
    mainDimension = namingParams["mainDimension"]
    varianceType = namingParams["varianceTypeName"]
    dfBase = dfDict[dfBaseKey]
    # Use project helper to duplicate Polars DataFrame/LazyFrame
    df = duplicate_dataframe(dfBase)
    group_byCols = [varianceType]
    if mainDimension in chartDict and len(chartDict[mainDimension]) > 0:
        group_byCols = chartDict[mainDimension] + group_byCols
    paramDict = get_data_sample(df, "before_mix_fix_dimension", False, paramDict)
    df = calculate_sales_mix_variance(df, paramDict, chartDict, indexCols)
    df, paramDict = calculate_margin_mix_variance(df, paramDict, chartDict, indexCols)
    paramDict = get_data_sample(df, "after_mix_fix_dimension", False, paramDict)
    df, paramDict = melt_data_on_variance_cols(df, paramDict, chartDict, indexCols)
    paramDict = get_data_sample(df, "after_melt_fix_dimension", False, paramDict)
    sumCols = [varianceAmount]
    return df, dfBase, paramDict, sumCols, group_byCols


def drop_zero_variance_rows(
    dfCopy, paramDict, chartDict, *, as_lazy: bool = False
) -> pl.DataFrame | pl.LazyFrame:
    """
    if a row has zero variance we drop it since it creates issues with driver variance calculation
    """
    namingParams = get_naming_params()
    totalVariance = namingParams["totalVariance"]
    cogsVariance = namingParams["COGSVariance"]
    discountVariance = namingParams["discountVariance"]
    driverAndUnitsAggregation = namingParams["driverAndUnitsAggregation"]
    driverAndVolumeAggregation = namingParams["driverAndVolumeAggregation"]
    mixUnitsDriverAggregation = namingParams["mixUnitsDriverAggregation"]
    mixVolumeDriverAggregation = namingParams["mixVolumeDriverAggregation"]
    varianceAggregation = namingParams["varianceAggregation"]
    if varianceAggregation in chartDict and chartDict[varianceAggregation] in [
        driverAndUnitsAggregation,
        driverAndVolumeAggregation,
        mixUnitsDriverAggregation,
        mixVolumeDriverAggregation,
    ]:
        input_is_lazy = isinstance(dfCopy, pl.LazyFrame)
        df = dfCopy.lazy() if not input_is_lazy else dfCopy
        columns, _ = get_schema_and_column_names(df)
        sum_cols = [
            c for c in [totalVariance, cogsVariance, discountVariance] if c in columns
        ]
        if sum_cols:
            df = df.filter(pl.sum_horizontal(sum_cols) != 0)
        if as_lazy or input_is_lazy:
            return df
        return df.collect()
    else:
        return ensure_lazyframe(dfCopy) if as_lazy else dfCopy


def exclude_outliers_from_mix_variance(
    df: pl.DataFrame | pl.LazyFrame, paramDict: dict
) -> pl.DataFrame | pl.LazyFrame:
    """
    Eliminate "extreme" prices from our mix variance calculation by excluding
    rows outside specified quantile thresholds. This version uses a lazy
    Polars pipeline.
    """
    namingParams = get_naming_params()
    configParams = get_config_params()
    periodsArray = configParams["periodsArray"]
    separatorString = namingParams["separatorString"]
    quantiles = namingParams["quantiles"]
    excludeOutliers = namingParams["excludeOutliers"]
    pricePerUnitName = namingParams["pricePerUnitName"]
    workColumn = namingParams["workColumn"]

    use_lazy = isinstance(df, pl.LazyFrame)
    lf = ensure_lazyframe(df)

    if excludeOutliers in paramDict and paramDict[excludeOutliers]:
        pricePeriodZero = pricePerUnitName + separatorString + periodsArray[0]
        pricePeriodOne = pricePerUnitName + separatorString + periodsArray[1]

        # Convert percentages into decimal quantiles
        min_quantile = paramDict[quantiles] / 100
        max_quantile = (100 - paramDict[quantiles]) / 100

        # 1) Add the work column (row-wise mean of the two prices)
        lf = lf.with_columns(
            ((pl.col(pricePeriodZero) + pl.col(pricePeriodOne)) / 2).alias(workColumn)
        )

        # 2) Compute the actual numerical cutoff values by selecting them into a separate LazyFrame
        q = lf.select(
            [
                pl.col(workColumn).quantile(min_quantile).alias("q_min"),
                pl.col(workColumn).quantile(max_quantile).alias("q_max"),
            ]
        )
        # (No computation happens yet; it's still lazy.)

        # 3) Cross join these values back to filter by them
        lf = (
            lf.join(q, how="cross")
            .filter(pl.col(workColumn).is_between(pl.col("q_min"), pl.col("q_max")))
            .drop([workColumn, "q_min", "q_max"])
        )
    return lf if use_lazy else lf.collect()


def count_number_of_nodes_in_index(
    df: pl.DataFrame | pl.LazyFrame, indexCols: list[str]
) -> pl.DataFrame | pl.LazyFrame:
    """
    since we can have equivalent rows (that filter the same things and that have the same number values)
    to choose the one to show we need to count the number of nodes in each row
    """
    namingParams = get_naming_params()
    numberOfNodes = namingParams["numberOfNodes"]
    maxNodes = len(indexCols)

    null_count = pl.sum_horizontal(
        [pl.col(col).is_null().cast(pl.Int64) for col in indexCols]
    )
    expr = (maxNodes - null_count).alias(numberOfNodes)

    return df.with_columns(expr)


def count_number_of_items_per_combination(
    df: pl.DataFrame | pl.LazyFrame, paramDict: dict, indexCols: list[str]
) -> pl.DataFrame | pl.LazyFrame:
    """
    we want to weigh more the combinations where columns contains more unique items (example product article) than those than contain less (example high medium low)
    """
    namingParams = get_naming_params()
    uniqueValuesInColumnDict = namingParams["uniqueValuesInColumnDict"]
    uniqueValuesInCombination = namingParams["uniqueValuesInCombination"]

    columns, _ = get_schema_and_column_names(df)
    weight_exprs = [
        pl.when(pl.col(col).is_null())
        .then(0)
        .otherwise(paramDict[uniqueValuesInColumnDict][col])
        for col in indexCols
        if col in columns and col in paramDict[uniqueValuesInColumnDict]
    ]

    combo_expr = pl.sum_horizontal(weight_exprs) if weight_exprs else pl.lit(0)
    df = df.with_columns(combo_expr.alias(uniqueValuesInCombination))
    df = df.with_row_index("_idx").drop("_idx")
    return df


def make_parent_key_df(dfCopy, nodeDict, paramDict):
    """
    we need to add back the parent information on
    all the groupedby and concatenated dataframe
    we pull the information about parents from the base dataframe
    extracting a df with no duplicates and only data dimension columns
    """
    runParams = get_run_params()
    deleteHierarchicalParents = runParams["deleteHierarchicalParents"]
    namingParams = get_naming_params()
    hierarchicalName = namingParams["hierarchicalName"]
    hierarchicalCols = []
    if deleteHierarchicalParents:
        if hierarchicalName in nodeDict and len(nodeDict[hierarchicalName]) > 0:
            df = ensure_polars_df(duplicate_dataframe(dfCopy))
            for element in nodeDict[hierarchicalName]:
                array = list(nodeDict[hierarchicalName][element].keys())
                hierarchicalCols = hierarchicalCols + array
            hierarchicalCols = unique(hierarchicalCols)
            # Project only the hierarchical columns using Polars select
            df = ensure_polars_df(df.select(hierarchicalCols))
            paramDict = check_if_duplicates_in_all_columns(df, "dfParents", paramDict)
            df = df.unique(maintain_order=True)
        else:
            df = pl.DataFrame()
    else:
        df = pl.DataFrame()
    return df, paramDict


def fill_parent_info(df, dfParentsCopy, nodeDict, paramDict):
    """Populate missing hierarchical parents using ``dfParentsCopy``."""

    runParams = get_run_params()
    deleteHierarchicalParents = runParams["deleteHierarchicalParents"]
    namingParams = get_naming_params()
    hierarchicalName = namingParams["hierarchicalName"]

    if deleteHierarchicalParents:
        df = ensure_polars_df(df)
        dfParentsCopy = ensure_polars_df(dfParentsCopy)

        pairs = [
            (rev[i], rev[i + 1])
            for element in nodeDict.get(hierarchicalName, {})
            for rev in [list(nodeDict[hierarchicalName][element].keys())[::-1]]
            for i in range(len(rev) - 1)
        ]

        lookups = [
            (
                child,
                parent,
                dfParentsCopy.select([child, parent]).unique(maintain_order=True),
            )
            for child, parent in pairs
        ]

        def join_lookup(acc_df, info):
            child, parent, lookup = info
            joined = acc_df.join(lookup, on=child, how="left", suffix="_parent")
            return joined.with_columns(
                pl.coalesce(pl.col(parent), pl.col(f"{parent}_parent")).alias(parent)
            ).drop(f"{parent}_parent")

        df = reduce(join_lookup, lookups, df)

        paramDict = check_if_duplicates_in_all_columns(
            df, "Fill parent info", paramDict
        )

    return df, paramDict


def get_and_process_index_combinations(df, indexCols, valueCols, nodeDict, paramDict):
    """
    grouping function together for order
    """
    namingParams = get_naming_params()
    dataPreparation = namingParams["dataPreparationName"]
    getIndexCombinations = namingParams["getIndexCombinationsName"]
    resultArray, paramDict = get_index_combinations(indexCols, nodeDict, paramDict)
    measure_time(dataPreparation, getIndexCombinations, False)
    df, paramDict = process_index_combinations(
        df, resultArray, indexCols, valueCols, paramDict
    )
    return df, paramDict


def make_df_copy_for_subtractions(dfCopy, indexCols, paramDict, chartDict):
    """
    while we need to identify equivalent node chains, so we do not get
    duplicate results, we need to keep these duplicate chains in the dataset
    because otherwise we would no do the correct subtractions where the
    subtraction joins is about the duplicate chains
    Here we create a copy
    If the dataframe is very big, we skip this step
    """
    namingParams = get_naming_params()
    configParams = get_config_params()
    numberOfColumnsWeight = configParams[namingParams["numberOfColumnsWeight"]]
    highEstimatorValue = configParams[namingParams["highEstimatorValue"]]
    estimator = (len(indexCols) - 1) * numberOfColumnsWeight * dfCopy.height
    simplifiedProcessing = namingParams["simplifiedProcessing"]
    metConditionValue = namingParams["metConditionValue"]
    notMetConditionValue = namingParams["notMetConditionValue"]
    if estimator < highEstimatorValue:
        paramDict[simplifiedProcessing] = notMetConditionValue
        df = duplicate_dataframe(dfCopy)
        df, paramDict = melt_data_on_variance_cols(df, paramDict, chartDict, indexCols)
        df, paramDict = change_nan_to_all(df, indexCols, paramDict)
    else:
        paramDict[simplifiedProcessing] = metConditionValue
        df = pl.DataFrame()
    return df, paramDict


def output_df_with_combinations(
    indexCols, nodeDict, df, valueCols, paramDict, chartDict
):
    """
    grouping function together for order
    """
    namingParams = get_naming_params()
    dataPreparation = namingParams["dataPreparationName"]
    makeParentKey = namingParams["makeParentKeyName"]
    fillParentInfo = namingParams["fillParentInfoName"]
    countNumberOfNodes = namingParams["countNumberOfNodesName"]
    countNumberOfNodesPerCombination = namingParams[
        "countNumberOfNodesPerCombinationName"
    ]
    makeCopyForSubtraction = namingParams["makeCopyForSubtractionName"]
    paramDict = get_data_sample(df, "output_df_with_combinations", False, paramDict)
    if len(indexCols) > 1 and paramDict[namingParams["numberOfPeriodsFound"]] > 1:
        dfParents, paramDict = make_parent_key_df(df, nodeDict, paramDict)
        measure_time(dataPreparation, makeParentKey, False)
        df, paramDictCopy = get_and_process_index_combinations(
            df, indexCols, valueCols, nodeDict, paramDict
        )
        paramDict = get_data_sample(
            df, "get_and_process_index_combinations", False, paramDict
        )
        df, paramDict = fill_parent_info(df, dfParents, nodeDict, paramDict)
        measure_time(dataPreparation, fillParentInfo, False)
        df = count_number_of_nodes_in_index(df, indexCols)
        measure_time(dataPreparation, countNumberOfNodes, False)
        df = count_number_of_items_per_combination(df, paramDict, indexCols)
        measure_time(dataPreparation, countNumberOfNodesPerCombination, False)
        dfSubtract, paramDict = make_df_copy_for_subtractions(
            df, indexCols, paramDict, chartDict
        )
        measure_time(dataPreparation, makeCopyForSubtraction, False)
    else:
        dfSubtract = pl.DataFrame()
    return df, dfSubtract, indexCols, paramDict


def output_combinations_and_melt_results(
    indexCols, nodeDict, df, valueCols, paramDict, chartDict
):
    """
    grouping function together for order
    """
    df, dfSubtract, indexCols, paramDict = output_df_with_combinations(
        indexCols, nodeDict, df, valueCols, paramDict, chartDict
    )
    if len(indexCols) > 1:
        paramDict = get_data_sample(
            df, "output_df_with_combinations_df", False, paramDict
        )
        paramDict = get_data_sample(
            df, "output_df_with_combinations_dfSubtract", False, paramDict
        )
        df, paramDict = delete_duplicate_nodes_and_melt_result(
            df, dfSubtract, indexCols, valueCols, paramDict, chartDict
        )
        df = filter_by_number_of_nodes(df, paramDict)
    return df, indexCols, paramDict


def merge_dataframe_with_subtraction_df(df, dfSubtract, indexCols, paramDict):
    """
    we need to add back the "duplicate" rows we took off because otherwise some
    subtractions will not work. We need to tag these duplicates so we do not
    get them as results. So we look for the combinations that are not in the main
    dataframe, tag them and add them back
    """
    namingParams = get_naming_params()
    randomKey = namingParams["randomKey"]
    metConditionValue = namingParams["metConditionValue"]
    notMetConditionValue = namingParams["notMetConditionValue"]
    workColumn = namingParams["workColumn"]
    simplifiedProcessing = namingParams["simplifiedProcessing"]
    simplifiedProcessingMessage = namingParams["simplifiedProcessingMessage"]
    if isinstance(df, pl.LazyFrame):
        row_count = df.select(pl.len()).collect()[0, 0]
    elif isinstance(df, pl.DataFrame):
        row_count = df.height
    else:
        row_count = df.height
    statusMessage = (
        "Dataset has **"
        + str("{:,.0f}".format(row_count))
        + "** combinations among which the results will be drawn. "
    )
    paramDict = add_status_message_to_paramDict(paramDict, statusMessage, 0)
    if paramDict[simplifiedProcessing]:
        paramDict = add_status_message_to_paramDict(
            paramDict, simplifiedProcessingMessage, 0
        )
    else:
        use_lazy = isinstance(df, pl.LazyFrame)
        df_base = ensure_lazyframe(df)
        dfFilter = df_base.select(indexCols).with_columns(
            pl.lit(metConditionValue).alias(workColumn)
        )
        dfSubtract = ensure_lazyframe(dfSubtract).join(
            dfFilter, on=indexCols, how="left"
        )
        dfSubtract = dfSubtract.filter(pl.col(workColumn) != metConditionValue)
        dfSubtract = drop_columns(dfSubtract, [workColumn])
        dfSubtract = dfSubtract.with_columns(pl.lit(0).alias(randomKey))
        # Use helper to retrieve column names consistently (avoids pandas-style APIs)
        columns, _ = get_schema_and_column_names(dfSubtract)
        columns = take_filtered_value_out_of_option_list(columns, randomKey)
        try:
            dfSubtract, paramDict = group_by_df_on_index_cols(
                dfSubtract, columns, [randomKey], "sum", paramDict, False
            )
            dfSubtract = dfSubtract.with_columns(pl.lit(None).alias(randomKey))
            target_schema = df_base.collect_schema()
            subtract_columns, _ = get_schema_and_column_names(dfSubtract)
            dfSubtract = dfSubtract.select(
                [
                    (
                        pl.col(column).cast(dtype, strict=False).alias(column)
                        if column in subtract_columns
                        else pl.lit(None, dtype=dtype).alias(column)
                    )
                    for column, dtype in target_schema.items()
                ]
            )
            df_base = pl.concat([df_base, dfSubtract], how="diagonal_relaxed")
            df = df_base if use_lazy else df_base.collect()
        except Exception as e:  # group_by may fail; fallback on simplified processing
            logging.exception(e)
            ui.error("Something went wrong with merge_dataframe_with_subtraction_df.")
            paramDict[simplifiedProcessing] = metConditionValue
            paramDict = add_status_message_to_paramDict(
                paramDict, simplifiedProcessingMessage, 7
            )
    return df, paramDict


def delete_duplicate_nodes_and_melt_result(
    df, dfSubtract, indexCols, valueCols, paramDict, chartDict
):
    """
    grouping function together for order
    """
    namingParams = get_naming_params()
    if len(indexCols) > 1 and paramDict[namingParams["numberOfPeriodsFound"]] > 1:
        df, paramDict = delete_duplicate_nodes(df, indexCols, valueCols, paramDict)
        paramDict = get_data_sample(df, "before_melt_without_mix", False, paramDict)
        df = calculate_sales_mix_variance(df, paramDict, chartDict, indexCols)
        df, paramDict = calculate_margin_mix_variance(
            df, paramDict, chartDict, indexCols
        )
        paramDict = get_data_sample(
            df, "with_mix_variance_calculation", False, paramDict
        )
        df, paramDict = melt_data_on_variance_cols(df, paramDict, chartDict, indexCols)
        paramDict = get_data_sample(df, "melt_data_on_variance_cols", False, paramDict)
        df, paramDict = change_nan_to_all(df, indexCols, paramDict)
        df, paramDict = merge_dataframe_with_subtraction_df(
            df, dfSubtract, indexCols, paramDict
        )
    return df, paramDict


def drop_not_required_variance_columns(df, toRemoveCols, valueNameArray):
    """
    based on the choice of the type of variance detail, we delete the useless columns
    """
    df = drop_columns(df, toRemoveCols)
    for column in toRemoveCols:
        if column in valueNameArray:
            valueNameArray = take_filtered_value_out_of_option_list(
                valueNameArray, column
            )
    return df, valueNameArray


def melt_data_on_variance_cols(df, paramDict, chartDict, indexCols):
    """
    using the "parents" df we want to populate the missing parent information in the
    core dataframe. We build a series of two column dictionaries and use them to map each parent column
    based on the info of the child
    """
    namingParams = get_naming_params()
    configParams = get_config_params()
    varianceAggregationParams = get_variance_aggregation_params()
    mixSalesAggregationArray = varianceAggregationParams[
        namingParams["mixSalesAggregationArray"]
    ]
    mixMarginAggregationArray = varianceAggregationParams[
        namingParams["mixMarginAggregationArray"]
    ]
    driverAggregationArray = varianceAggregationParams[
        namingParams["driverAggregationArray"]
    ]
    discountName = namingParams["discountName"]
    priceVariance = namingParams["priceVariance"]
    volumeVariance = namingParams["volumeVariance"]
    changedVolumeVariance = namingParams["changedVolumeVarianceMixName"]
    lostVolumeVariance = namingParams["lostVolumeVarianceName"]
    newVolumeVariance = namingParams["newVolumeVarianceName"]
    discountVariance = namingParams["discountVariance"]
    COGSVariance = namingParams["COGSVariance"]
    totalVariance = namingParams["totalVariance"]
    mixVariance = namingParams["mixVariance"]
    varianceStemArray = configParams["varianceStemArray"]
    varianceTypeName = namingParams["varianceTypeName"]
    varianceAmountName = namingParams["varianceAmountName"]
    simplifiedProcessing = namingParams["simplifiedProcessing"]
    varianceAggregation = namingParams["varianceAggregation"]
    discountsUnitsCogsAggregation = namingParams["discountsUnitsCogsAggregation"]
    discountsVolumeCogsAggregation = namingParams["discountsVolumeCogsAggregation"]
    totalVarianceAggregation = namingParams["totalVarianceAggregation"]
    priceAndUnitsAggregation = namingParams["priceAndUnitsAggregation"]
    priceAndVolumeAggregation = namingParams["priceAndVolumeAggregation"]
    unitPriceOnSalesAggregation = namingParams["unitPriceOnSalesAggregation"]
    volumePriceOnSalesAggregation = namingParams["volumePriceOnSalesAggregation"]
    unitsOnSalesAggregation = namingParams["unitsOnSalesAggregation"]
    volumeOnSalesAggregation = namingParams["volumeOnSalesAggregation"]
    newAndLostUnitsAggregation = namingParams["newAndLostUnitsAggregation"]
    newAndLostVolumeAggregation = namingParams["newAndLostVolumeAggregation"]
    newAndLostUnitsMixAggregation = namingParams["newAndLostUnitsMixAggregation"]
    newAndLostVolumeMixAggregation = namingParams["newAndLostVolumeMixAggregation"]
    unitPriceOnMarginAggregation = namingParams["unitPriceOnMarginAggregation"]
    volumePriceOnMarginAggregation = namingParams["volumePriceOnMarginAggregation"]
    unitsOnMarginAggregation = namingParams["unitsOnMarginAggregation"]
    volumeOnMarginAggregation = namingParams["volumeOnMarginAggregation"]
    newAggregation = namingParams["newAggregation"]
    lostAggregation = namingParams["lostAggregation"]
    changedAggregation = namingParams["changedAggregation"]
    marginUnitsRateAggregation = namingParams["marginUnitsRateAggregation"]
    marginVolumeRateAggregation = namingParams["marginVolumeRateAggregation"]
    costsUnitsAggregation = namingParams["costsUnitsAggregation"]
    costsVolumeAggregation = namingParams["costsVolumeAggregation"]
    costsUnitsMixAggregation = namingParams["costsUnitsMixAggregation"]
    costsVolumeMixAggregation = namingParams["costsVolumeMixAggregation"]
    discountsAndUnitsAggregation = namingParams["discountsAndUnitsAggregation"]
    discountsAndVolumeAggregation = namingParams["discountsAndVolumeAggregation"]
    discountsAggregation = namingParams["discountsAggregation"]
    cogsAggregation = namingParams["cogsAggregation"]
    driverAndUnitsAggregation = namingParams["driverAndUnitsAggregation"]
    driverAndVolumeAggregation = namingParams["driverAndVolumeAggregation"]
    driverAggregation = namingParams["driverAggregation"]
    mixAggregation = namingParams["mixAggregation"]
    metConditionValue = namingParams["metConditionValue"]
    notMetConditionValue = namingParams["notMetConditionValue"]
    netOfDiscountVariance = namingParams["netOfDiscountVariance"]
    marginVariance = namingParams["marginVariance"]
    percentSuffix = namingParams["percentSuffix"]
    netOfDiscountAggregation = namingParams["netOfDiscountAggregation"]
    marginVarianceAggregation = namingParams["marginVarianceAggregation"]
    marginRateVariance = namingParams["marginRateVariance"]
    pureVolumeVariance = namingParams["pureVolumeVarianceName"]
    costVariance = namingParams["costVariance"]
    workColumn = namingParams["workColumn"]
    varianceInPercent = namingParams["varianceInPercent"]
    percentSuffix = namingParams["percentSuffix"]
    pureVolumeVarianceName = namingParams["pureVolumeVarianceName"]
    mixUnitsDriverAggregation = namingParams["mixUnitsDriverAggregation"]
    mixVolumeDriverAggregation = namingParams["mixVolumeDriverAggregation"]
    categoryWeightedDistributionName = namingParams["categoryWeightedDistributionName"]
    volumePerPosVariance = namingParams["volumePerPosVariance"]
    mixVolumePerPosVariance = namingParams["mixVolumePerPosVariance"]
    averageTicketVariance = namingParams["averageTicketVariance"]
    mixAverageTicketVariance = namingParams["mixAverageTicketVariance"]
    marginVariancePercent = marginVariance + percentSuffix
    netOfDiscountVariancePercent = netOfDiscountVariance + percentSuffix
    cogsVariancePercent = COGSVariance + percentSuffix
    volumeVariancePercent = volumeVariance + percentSuffix
    priceVariancePercent = priceVariance + percentSuffix
    costVariancePercent = costVariance + percentSuffix
    marginRateVariancePercent = marginRateVariance + percentSuffix
    percentAggregations = [
        marginVariancePercent,
        netOfDiscountVariancePercent,
        cogsVariancePercent,
        volumeVariancePercent,
        priceVariancePercent,
        costVariancePercent,
        marginRateVariancePercent,
    ]
    idVarArray = []
    valueNameArray = []
    columns, schema = get_schema_and_column_names(df)
    for column in columns:
        isVariance = False
        if column in varianceStemArray:
            isVariance = True
        if isVariance:
            if column not in valueNameArray:
                valueNameArray.append(column)
        # elif percentSuffix not in column:
        elif column not in idVarArray:
            idVarArray.append(column)
    toRemoveCols = []
    if (
        chartDict[varianceAggregation]
        in [
            priceAndUnitsAggregation,
            priceAndVolumeAggregation,
            newAndLostUnitsAggregation,
            newAndLostVolumeAggregation,
            newAndLostUnitsMixAggregation,
            newAndLostVolumeMixAggregation,
            newAggregation,
            lostAggregation,
            changedAggregation,
            driverAndUnitsAggregation,
            driverAndVolumeAggregation,
            driverAggregation,
            unitPriceOnSalesAggregation,
            volumePriceOnSalesAggregation,
            unitsOnSalesAggregation,
            volumeOnSalesAggregation,
        ]
        + mixSalesAggregationArray
    ):
        toRemoveCols = [
            totalVariance,
            discountVariance,
            COGSVariance,
            marginRateVariance,
            costVariance,
        ]
        df, valueNameArray = drop_not_required_variance_columns(
            df, toRemoveCols, valueNameArray
        )
        if chartDict[varianceAggregation] in [
            unitPriceOnSalesAggregation,
            volumePriceOnSalesAggregation,
        ]:
            toRemoveCols = [volumeVariance]
            df, valueNameArray = drop_not_required_variance_columns(
                df, toRemoveCols, valueNameArray
            )
        elif chartDict[varianceAggregation] in [
            unitsOnSalesAggregation,
            volumeOnSalesAggregation,
        ]:
            toRemoveCols = [priceVariance]
            df, valueNameArray = drop_not_required_variance_columns(
                df, toRemoveCols, valueNameArray
            )
        elif chartDict[varianceAggregation] in [newAggregation]:
            toRemoveCols = [priceVariance, lostVolumeVariance, changedVolumeVariance]
            df, valueNameArray = drop_not_required_variance_columns(
                df, toRemoveCols, valueNameArray
            )
        elif chartDict[varianceAggregation] in [lostAggregation]:
            toRemoveCols = [priceVariance, newVolumeVariance, changedVolumeVariance]
            df, valueNameArray = drop_not_required_variance_columns(
                df, toRemoveCols, valueNameArray
            )
        elif chartDict[varianceAggregation] in [changedAggregation]:
            toRemoveCols = [priceVariance, lostVolumeVariance, newVolumeVariance]
            df, valueNameArray = drop_not_required_variance_columns(
                df, toRemoveCols, valueNameArray
            )
        elif chartDict[varianceAggregation] in [driverAggregation]:
            toRemoveCols = [priceVariance, volumeVariance] + mixSalesAggregationArray
            df, valueNameArray = drop_not_required_variance_columns(
                df, toRemoveCols, valueNameArray
            )
        elif chartDict[varianceAggregation] in [mixAggregation]:
            toRemoveCols = [priceVariance, pureVolumeVariance] + driverAggregationArray
            df, valueNameArray = drop_not_required_variance_columns(
                df, toRemoveCols, valueNameArray
            )
        elif chartDict[varianceAggregation] in [
            driverAndUnitsAggregation,
            driverAndVolumeAggregation,
        ]:
            if categoryWeightedDistributionName in columns:
                df = df.rename({volumeVariance: mixVolumePerPosVariance})
            else:
                df = df.rename({volumeVariance: mixAverageTicketVariance})
        elif chartDict[varianceAggregation] in [
            mixUnitsDriverAggregation,
            mixVolumeDriverAggregation,
        ]:
            if categoryWeightedDistributionName in columns:
                df = df.rename({pureVolumeVariance: volumePerPosVariance})
            else:
                df = df.rename({pureVolumeVariance: averageTicketVariance})
    elif chartDict[varianceAggregation] in [
        totalVarianceAggregation,
    ]:
        toRemoveCols = [
            priceVariance,
            volumeVariance,
            lostVolumeVariance,
            newVolumeVariance,
            mixVariance,
            discountVariance,
            COGSVariance,
            marginRateVariance,
            costVariance,
        ]
        df, valueNameArray = drop_not_required_variance_columns(
            df, toRemoveCols, valueNameArray
        )
    elif chartDict[varianceAggregation] in [
        discountsUnitsCogsAggregation,
        discountsVolumeCogsAggregation,
        discountsAggregation,
        cogsAggregation,
        unitPriceOnMarginAggregation,
        volumePriceOnMarginAggregation,
        unitsOnMarginAggregation,
        volumeOnMarginAggregation,
    ]:
        toRemoveCols = [
            totalVariance,
            lostVolumeVariance,
            newVolumeVariance,
            mixVariance,
            marginRateVariance,
            costVariance,
        ]
        df, valueNameArray = drop_not_required_variance_columns(
            df, toRemoveCols, valueNameArray
        )
        if chartDict[varianceAggregation] in [discountsAggregation]:
            toRemoveCols = [priceVariance, volumeVariance, COGSVariance]
            df, valueNameArray = drop_not_required_variance_columns(
                df, toRemoveCols, valueNameArray
            )
        elif chartDict[varianceAggregation] in [cogsAggregation]:
            toRemoveCols = [priceVariance, volumeVariance, discountVariance]
            df, valueNameArray = drop_not_required_variance_columns(
                df, toRemoveCols, valueNameArray
            )
        elif chartDict[varianceAggregation] in [
            unitPriceOnMarginAggregation,
            volumePriceOnMarginAggregation,
        ]:
            toRemoveCols = [COGSVariance, volumeVariance, discountVariance]
            df, valueNameArray = drop_not_required_variance_columns(
                df, toRemoveCols, valueNameArray
            )
        elif chartDict[varianceAggregation] in [
            unitsOnMarginAggregation,
            volumeOnMarginAggregation,
        ]:
            toRemoveCols = [priceVariance, COGSVariance, discountVariance]
            df, valueNameArray = drop_not_required_variance_columns(
                df, toRemoveCols, valueNameArray
            )
    elif chartDict[varianceAggregation] in [
        discountsAndUnitsAggregation,
        discountsAndVolumeAggregation,
    ]:
        toRemoveCols = [
            totalVariance,
            lostVolumeVariance,
            newVolumeVariance,
            mixVariance,
            COGSVariance,
            marginRateVariance,
        ]
        df, valueNameArray = drop_not_required_variance_columns(
            df, toRemoveCols, valueNameArray
        )
    elif chartDict[varianceAggregation] in [
        marginUnitsRateAggregation,
        marginVolumeRateAggregation,
    ]:
        toRemoveCols = [
            totalVariance,
            lostVolumeVariance,
            newVolumeVariance,
            mixVariance,
            discountVariance,
            COGSVariance,
            costVariance,
        ]
        df, valueNameArray = drop_not_required_variance_columns(
            df, toRemoveCols, valueNameArray
        )
    elif chartDict[varianceAggregation] in [
        costsUnitsAggregation,
        costsVolumeAggregation,
    ]:
        toRemoveCols = [
            totalVariance,
            lostVolumeVariance,
            newVolumeVariance,
            discountVariance,
            COGSVariance,
            marginRateVariance,
            mixVariance,
        ]
        df, valueNameArray = drop_not_required_variance_columns(
            df, toRemoveCols, valueNameArray
        )
    elif chartDict[varianceAggregation] in [
        costsUnitsMixAggregation,
        costsVolumeMixAggregation,
    ]:
        toRemoveCols = [
            totalVariance,
            lostVolumeVariance,
            newVolumeVariance,
            discountVariance,
            COGSVariance,
            marginRateVariance,
        ]
        df, valueNameArray = drop_not_required_variance_columns(
            df, toRemoveCols, valueNameArray
        )
    elif chartDict[varianceAggregation] in [marginVarianceAggregation]:
        if chartDict[varianceInPercent] != metConditionValue:
            present_cols = [
                c
                for c in [totalVariance, discountVariance, COGSVariance]
                if c in columns
            ]
            if present_cols:
                df = df.with_columns(
                    pl.sum_horizontal([pl.col(c) for c in present_cols]).alias(
                        marginVariance
                    )
                )
        elif chartDict[varianceInPercent] == metConditionValue:
            toRemoveCols = [netOfDiscountVariancePercent]
            df, paramDict = calculate_variance_in_percent(
                df, paramDict, chartDict, False
            )
        toRemoveCols = [
            totalVariance,
            priceVariance,
            volumeVariance,
            lostVolumeVariance,
            newVolumeVariance,
            mixVariance,
            discountVariance,
            COGSVariance,
            marginRateVariance,
            costVariance,
        ] + toRemoveCols
        df, valueNameArray = drop_not_required_variance_columns(
            df, toRemoveCols, valueNameArray
        )
    elif (
        chartDict[varianceAggregation] in [netOfDiscountAggregation]
        and discountName in columns
    ):
        if chartDict[varianceInPercent] != metConditionValue:
            df = df.with_columns(
                (pl.col(totalVariance) + pl.col(discountVariance)).alias(
                    netOfDiscountVariance
                )
            )
        toRemoveCols = [
            totalVariance,
            priceVariance,
            volumeVariance,
            lostVolumeVariance,
            newVolumeVariance,
            mixVariance,
            discountVariance,
            COGSVariance,
            marginRateVariance,
            costVariance,
        ] + toRemoveCols
        df, valueNameArray = drop_not_required_variance_columns(
            df, toRemoveCols, valueNameArray
        )
    if chartDict[varianceInPercent] != metConditionValue:
        toRemoveCols = percentAggregations
        df, valueNameArray = drop_not_required_variance_columns(
            df, toRemoveCols, valueNameArray
        )
    df = df.unpivot(
        index=idVarArray,
        variable_name=varianceTypeName,
        value_name=varianceAmountName,
    )
    df = df.sort(by=varianceAmountName, descending=True)
    if simplifiedProcessing in paramDict and paramDict[simplifiedProcessing]:
        df = ensure_polars_df(df)
        df = df.filter(pl.col(varianceAmountName) != 0)
    df = df.sort(by=varianceAmountName, descending=True)
    df = round_value_columns_to_dec(df, [varianceAmountName])
    divideArray = make_divideArray(df, indexCols)
    df, paramDict = divide_back_if_multiplied(
        df, paramDict, chartDict, divideArray, False
    )
    return df, paramDict


def calculate_aggregated_price_volume_variance_value(
    df, paramDict, periodsArray, columns, chartDict
):
    """
    if a discount columns exist, calculates the total variance after discount
    """
    namingParams = get_naming_params()
    configParams = get_config_params()
    amountName = namingParams["monetaryLocalCurrencyName"]
    separatorString = namingParams["separatorString"]
    totalVariance = namingParams["totalVariance"]
    varianceName = namingParams["varianceName"]
    selectedPeriods = namingParams["selectedPeriods"]
    periodName = namingParams["periodName"]
    totalVarianceValueKey = namingParams["totalVarianceValue"]
    totalAmountPeriodZeroKey = namingParams["totalAmountPeriodZero"]
    totalAmountPeriodOneKey = namingParams["totalAmountPeriodOne"]
    avgAmountPeriodsZeroOneKey = namingParams["avgAmountPeriodsZeroOne"]
    varianceOnTotal = namingParams["varianceOnTotal"]
    yoyPercentChange = namingParams["yoyPercentChange"]
    nothingThereString = namingParams["nothingThereString"]
    firstPeriodSales = namingParams["firstPeriodSales"]
    secondPeriodSales = namingParams["secondPeriodSales"]
    multiplyConstant = configParams[namingParams["multiplyConstant"]]
    amountPeriodZero = amountName + separatorString + periodsArray[0]
    amountPeriodOne = amountName + separatorString + periodsArray[1]
    varianceSum, periodZeroSum = 0, 0
    periodOneSum, periodZeroOneAverage = 0, 0
    if amountPeriodZero in columns:
        periodZeroSum = df[amountPeriodZero].sum()
        periodOneSum = df[amountPeriodOne].sum()
        if totalVariance in columns:
            varianceSum = df[totalVariance].sum()
        else:
            varianceSum = periodOneSum - periodZeroSum
    else:
        periods = chartDict[selectedPeriods]
        df_temp = ensure_polars_df(df)
        periodZeroSum = df_temp.filter(pl.col(periodName) == periods[0])[
            amountName
        ].sum()
        periodOneSum = df_temp.filter(pl.col(periodName) == periods[1])[
            amountName
        ].sum()
        varianceSum = periodOneSum - periodZeroSum
    periodZeroOneAverage = (periodOneSum + periodZeroSum) / 2
    periodZeroSum, periodOneSum = round(periodZeroSum, 2), round(periodOneSum, 2)
    periodZeroOneAverage, varianceSum = (
        round(periodZeroOneAverage, 2),
        round(varianceSum, 2),
    )
    if paramDict[namingParams["isColumnMultiplied"]]:
        varianceSum = round(varianceSum / multiplyConstant, 2)
        periodZeroOneAverage = round(periodZeroOneAverage / multiplyConstant, 2)
        periodZeroSum = round(periodZeroSum / multiplyConstant, 2)
        periodOneSum = round(periodOneSum / multiplyConstant, 2)
    paramDict[totalVarianceValueKey], paramDict[totalAmountPeriodZeroKey] = (
        varianceSum,
        periodZeroSum,
    )
    paramDict[totalAmountPeriodOneKey], paramDict[avgAmountPeriodsZeroOneKey] = (
        periodOneSum,
        periodZeroOneAverage,
    )
    if abs(varianceSum) > 100:
        varianceValue = str("{:,.0f}".format(round(varianceSum, 0)))
    else:
        varianceValue = str("{:,.1f}".format(round(varianceSum, 1)))
    if periodZeroOneAverage != 0:
        varianceOnTotalValue = round((varianceSum / periodZeroOneAverage) * 100, 1)
        if periodZeroSum != 0:
            yoy_delta = (periodOneSum * 100 / periodZeroSum) - 100
            if abs(yoy_delta) > 100:
                yoyPercentChangeValue = str("{:,.0f}".format(round(yoy_delta, 0)))
            else:
                yoyPercentChangeValue = str("{:,.1f}".format(round(yoy_delta, 1)))
        else:
            yoyPercentChangeValue = nothingThereString
        statusMessage = [
            totalVarianceValueKey + " " + varianceName,
            varianceValue,
            yoyPercentChangeValue,
        ]
        statusMessage = check_if_reverse_KPI_color(
            paramDict, statusMessage, amountName, periodZeroSum, periodOneSum
        )
        paramDict = add_status_message_to_paramDict(paramDict, statusMessage, 3)
        paramDict[yoyPercentChange] = yoyPercentChangeValue
        paramDict[varianceOnTotal] = varianceOnTotalValue
        # paramDict[varianceOnTotal]=yoyPercentChangeValue # to adjust parameters we could use either the change on the average market size or the change as % on the two periods
    if abs(periodZeroSum) > 100:
        printValue = str("{:,.0f}".format(round(periodZeroSum, 0)))
    else:
        printValue = str("{:,.1f}".format(round(periodZeroSum, 1)))
    statusMessage = "1st period: **" + printValue + "**."
    paramDict[firstPeriodSales] = statusMessage
    paramDict = add_status_message_to_paramDict(paramDict, statusMessage, 4)
    if abs(periodOneSum) > 100:
        printValue = str("{:,.0f}".format(round(periodOneSum, 0)))
    else:
        printValue = str("{:,.1f}".format(round(periodOneSum, 1)))
    statusMessage = " 2nd period: **" + printValue + "**."
    paramDict[secondPeriodSales] = statusMessage
    paramDict = add_status_message_to_paramDict(paramDict, statusMessage, 4)
    return paramDict


def check_if_reverse_KPI_color(
    paramDict, statusMessage, metric, valuePeriodZero, valuePeriodOne
):
    """
    checks if the color of the % change value in the ui.metric widget must be reversed from red to green to green to red
    """
    namingParams = get_naming_params()
    colorChoice = namingParams["colorChoice"]
    redToGreen = namingParams["redToGreen"]
    discountName = namingParams["discountName"]
    indirectCostsName = namingParams["indirectCostsName"]
    cogsName = namingParams["cogsName"]
    deltaColor = namingParams["deltaColor"]
    reverseColorMetricsArray = [discountName, indirectCostsName, cogsName]
    deltaColorValue = 1
    if valuePeriodZero < 0 and valuePeriodOne < 0:
        deltaColorValue = deltaColorValue * -1
    if colorChoice in paramDict and paramDict[colorChoice] not in [redToGreen]:
        deltaColorValue = deltaColorValue * -1
    if metric in reverseColorMetricsArray:
        deltaColorValue = deltaColorValue * -1
    if deltaColorValue == 1:
        deltaColor = "normal"
    else:
        deltaColor = "inverse"
    statusMessage.append(deltaColor)
    return statusMessage


def calculate_aggregated_discount_variance_value(df, paramDict, periodsArray, columns):
    """
    if a discount columns exist, calculates the total variance after discount
    """
    namingParams = get_naming_params()
    configParams = get_config_params()
    discountVariance = namingParams["discountVariance"]
    netOfDiscount = namingParams["netOfDiscountName"]
    varianceName = namingParams["varianceName"]
    separatorString = namingParams["separatorString"]
    netOfDiscountVarianceKey = namingParams["netOfDiscountVariance"]
    totalAmountPeriodZero = paramDict[namingParams["totalAmountPeriodZero"]]
    totalAmountPeriodOne = paramDict[namingParams["totalAmountPeriodOne"]]
    totalNetOfDiscountPeriodZero = namingParams["totalNetOfDiscountPeriodZero"]
    totalNetOfDiscountPeriodOne = namingParams["totalNetOfDiscountPeriodOne"]
    avgNetOfDiscountPeriodsZeroOne = namingParams["avgNetOfDiscountPeriodsZeroOne"]
    totalNetOfDiscountPeriodZeroinPercentKey = namingParams[
        "totalNetOfDiscountPeriodZeroinPercent"
    ]
    totalNetOfDiscountPeriodOneinPercentKey = namingParams[
        "totalNetOfDiscountPeriodOneinPercent"
    ]
    firstPeriodDiscountsKey = namingParams["firstPeriodDiscounts"]
    secondPeriodDiscountsKey = namingParams["secondPeriodDiscounts"]
    multiplyConstant = configParams[namingParams["multiplyConstant"]]
    if discountVariance in columns:
        netOfDiscountPeriodZero = netOfDiscount + separatorString + periodsArray[0]
        netOfDiscountPeriodOne = netOfDiscount + separatorString + periodsArray[1]
        periodZeroNetOfDiscountSum = df[netOfDiscountPeriodZero].sum()
        periodOneNetOfDiscountSum = df[netOfDiscountPeriodOne].sum()
        firstPeriodDiscounts = totalAmountPeriodZero - periodZeroNetOfDiscountSum
        secondPeriodDiscounts = totalAmountPeriodOne - periodOneNetOfDiscountSum
        firstPeriodDiscounts = str("{:,.0f}".format(round(firstPeriodDiscounts, 0)))
        secondPeriodDiscounts = str("{:,.0f}".format(round(secondPeriodDiscounts, 0)))
        paramDict[firstPeriodDiscountsKey] = (
            " 1st period: **" + firstPeriodDiscounts + "**."
        )
        paramDict[secondPeriodDiscountsKey] = (
            " 2nd period: **" + secondPeriodDiscounts + "**."
        )
        averageNetDiscounts = (
            periodZeroNetOfDiscountSum + periodOneNetOfDiscountSum
        ) / 2
        discountVarianceSum = periodOneNetOfDiscountSum - periodZeroNetOfDiscountSum
        periodOnePercentMarginAfterDiscount, periodZeroPercentMarginAfterDiscount = 0, 0
        if paramDict[namingParams["isColumnMultiplied"]]:
            discountVarianceSum = discountVarianceSum / multiplyConstant
            averageNetDiscounts = averageNetDiscounts / multiplyConstant
            periodZeroNetOfDiscountSum = periodZeroNetOfDiscountSum / multiplyConstant
            periodOneNetOfDiscountSum = periodOneNetOfDiscountSum / multiplyConstant
        paramDict[netOfDiscountVarianceKey] = discountVarianceSum
        paramDict[avgNetOfDiscountPeriodsZeroOne] = averageNetDiscounts
        if totalAmountPeriodZero != 0:
            periodZeroPercentMarginAfterDiscount = (
                periodZeroNetOfDiscountSum / totalAmountPeriodZero * 100
            )
            paramDict[totalNetOfDiscountPeriodZeroinPercentKey] = (
                periodZeroPercentMarginAfterDiscount
            )
        if totalAmountPeriodOne != 0:
            periodOnePercentMarginAfterDiscount = (
                periodOneNetOfDiscountSum / totalAmountPeriodOne * 100
            )
            paramDict[totalNetOfDiscountPeriodOneinPercentKey] = (
                periodOnePercentMarginAfterDiscount
            )
        percentMarginAfterdiscountUnitChange = (
            periodOnePercentMarginAfterDiscount - periodZeroPercentMarginAfterDiscount
        )
        paramDict[namingParams["percentVarianceAfterDiscounts"]] = (
            percentMarginAfterdiscountUnitChange
        )
        (
            paramDict[totalNetOfDiscountPeriodZero],
            paramDict[totalNetOfDiscountPeriodOne],
        ) = (periodZeroNetOfDiscountSum, periodOneNetOfDiscountSum)
        if abs(discountVarianceSum) > 100:
            discountVarianceValue = str("{:,.0f}".format(round(discountVarianceSum, 0)))
        else:
            discountVarianceValue = str("{:,.1f}".format(round(discountVarianceSum, 1)))
        statusMessage = (
            netOfDiscountVarianceKey + ": **" + discountVarianceValue + "**. "
        )
        if periodZeroNetOfDiscountSum != 0:
            yoyPercentChangeDelta = (
                periodOneNetOfDiscountSum * 100 / periodZeroNetOfDiscountSum
            ) - 100
            if abs(discountVarianceSum) > 100:
                yoyPercentChangeValue = str(
                    "{:,.1f}".format(round(yoyPercentChangeDelta, 1))
                )
            else:
                yoyPercentChangeValue = str(
                    "{:,.1f}".format(round(percentMarginAfterdiscountUnitChange, 1))
                )
        else:
            yoyPercentChangeValue = nothingThereString
        statusMessage = [
            netOfDiscountVarianceKey + " " + varianceName,
            discountVarianceValue,
            yoyPercentChangeValue,
        ]
        statusMessage = check_if_reverse_KPI_color(
            paramDict,
            statusMessage,
            netOfDiscountVarianceKey,
            periodZeroNetOfDiscountSum,
            periodOneNetOfDiscountSum,
        )
        paramDict = add_status_message_to_paramDict(paramDict, statusMessage, 5)
        if abs(discountVarianceSum) > 100:
            printValue = str("{:,.0f}".format(round(periodZeroNetOfDiscountSum, 0)))
        else:
            printValue = str("{:,.1f}".format(round(periodZeroNetOfDiscountSum, 1)))
        statusMessage1stA = "1st period after discount: **" + printValue + "**. "
        if abs(discountVarianceSum) > 100:
            printValue = str("{:,.0f}".format(round(periodOneNetOfDiscountSum, 0)))
        else:
            printValue = str("{:,.1f}".format(round(periodOneNetOfDiscountSum, 1)))
        statusMessage2ndA = "2nd period after discount: **" + printValue + "**. "
        if abs(discountVarianceSum) > 100:
            printValue = str(
                "{:,.1f}".format(round(periodZeroPercentMarginAfterDiscount, 1))
            )
        else:
            printValue = str(
                "{:,.1f}".format(round(periodZeroPercentMarginAfterDiscount, 1))
            )
        statusMessage1stB = (
            "1st period after discount as % of revenues: **" + printValue + "%**. "
        )
        if abs(discountVarianceSum) > 100:
            printValue = str(
                "{:,.1f}".format(round(periodOnePercentMarginAfterDiscount, 1))
            )
        else:
            printValue = str(
                "{:,.1f}".format(round(periodOnePercentMarginAfterDiscount, 1))
            )
        statusMessage2ndB = (
            "2nd period after discount as % of revenues: **" + printValue + "%**. "
        )
        paramDict = add_status_message_to_paramDict(
            paramDict, statusMessage1stA + " " + statusMessage1stB, 6
        )
        paramDict = add_status_message_to_paramDict(
            paramDict, statusMessage2ndA + " " + statusMessage2ndB, 6
        )
    return paramDict


def calculate_aggregated_cogs_variance_value(df, paramDict, periodsArray, columns):
    """
    if a discount columns exist, calculates the total variance after discount
    """
    namingParams = get_naming_params()
    configParams = get_config_params()
    COGSVariance = namingParams["COGSVariance"]
    varianceName = namingParams["varianceName"]
    marginName = namingParams["marginName"]
    marginVarianceKey = namingParams["marginVariance"]
    separatorString = namingParams["separatorString"]
    percentVarianceAfterCogs = namingParams["percentVarianceAfterCogs"]
    totalAmountPeriodZero = paramDict[namingParams["totalAmountPeriodZero"]]
    totalAmountPeriodOne = paramDict[namingParams["totalAmountPeriodOne"]]
    totalMarginPeriodZeroKey = namingParams["totalMarginPeriodZero"]
    totalMarginPeriodOneKey = namingParams["totalMarginPeriodOne"]
    totalMarginPeriodZeroinPercent = namingParams["totalMarginPeriodZeroinPercent"]
    totalMarginPeriodOneinPercent = namingParams["totalMarginPeriodOneinPercent"]
    firstPeriodMargin = namingParams["firstPeriodMargin"]
    secondPeriodMargin = namingParams["secondPeriodMargin"]
    firstPeriodDiscounts = namingParams["firstPeriodDiscounts"]
    secondPeriodDiscounts = namingParams["secondPeriodDiscounts"]
    avgMarginPeriodsZeroOne = namingParams["avgMarginPeriodsZeroOne"]
    multiplyConstant = configParams[namingParams["multiplyConstant"]]
    periodOnePercentMarginAfterCOGS, periodZeroPercentMarginAfterCOGS = 0, 0
    if COGSVariance in columns:
        marginZero = marginName + separatorString + periodsArray[0]
        marginOne = marginName + separatorString + periodsArray[1]
        periodZeroMarginSum = df[marginZero].sum()
        periodOneMarginSum = df[marginOne].sum()
        averageMargin = (periodZeroMarginSum + periodOneMarginSum) / 2
        marginVarianceSum = periodOneMarginSum - periodZeroMarginSum
        if paramDict[namingParams["isColumnMultiplied"]]:
            marginVarianceSum = marginVarianceSum / multiplyConstant
            averageMargin = averageMargin / multiplyConstant
            periodZeroMarginSum = periodZeroMarginSum / multiplyConstant
            periodOneMarginSum = periodOneMarginSum / multiplyConstant
        if totalAmountPeriodZero != 0:
            periodZeroPercentMarginAfterCOGS = (
                periodZeroMarginSum / totalAmountPeriodZero * 100
            )
            paramDict[totalMarginPeriodZeroinPercent] = periodZeroPercentMarginAfterCOGS
        if totalAmountPeriodOne != 0:
            periodOnePercentMarginAfterCOGS = (
                periodOneMarginSum / totalAmountPeriodOne * 100
            )
            paramDict[totalMarginPeriodOneinPercent] = periodOnePercentMarginAfterCOGS
        percentMarginAftercogsUnitChange = (
            periodOnePercentMarginAfterCOGS - periodZeroPercentMarginAfterCOGS
        )
        paramDict[percentVarianceAfterCogs] = percentMarginAftercogsUnitChange
        paramDict[marginVarianceKey] = marginVarianceSum
        paramDict[avgMarginPeriodsZeroOne] = averageMargin
        paramDict[totalMarginPeriodZeroKey], paramDict[totalMarginPeriodOneKey] = (
            periodZeroMarginSum,
            periodOneMarginSum,
        )
        if abs(marginVarianceSum) > 100:
            marginVarianceValue = str("{:,.0f}".format(round(marginVarianceSum, 0)))
        else:
            marginVarianceValue = str("{:,.1f}".format(round(marginVarianceSum, 1)))
        if periodZeroMarginSum != 0:
            yoyPercentChangeDelta = (
                periodOneMarginSum * 100 / periodZeroMarginSum
            ) - 100
            if abs(marginVarianceSum) > 100:
                yoyPercentChangeValue = str(
                    "{:,.1f}".format(round(yoyPercentChangeDelta, 1))
                )
            else:
                yoyPercentChangeValue = str(
                    "{:,.1f}".format(round(yoyPercentChangeDelta, 1))
                )
        else:
            yoyPercentChangeValue = nothingThereString
        statusMessage = [
            marginVarianceKey + " " + varianceName,
            marginVarianceValue,
            yoyPercentChangeValue,
        ]
        statusMessage = check_if_reverse_KPI_color(
            paramDict,
            statusMessage,
            marginName,
            periodZeroMarginSum,
            periodOneMarginSum,
        )
        paramDict = add_status_message_to_paramDict(paramDict, statusMessage, 3)
        if abs(marginVarianceSum) > 100:
            printValue = str("{:,.0f}".format(round(periodZeroMarginSum, 0)))
        else:
            printValue = str("{:,.1f}".format(round(periodZeroMarginSum, 1)))
        statusMessage1stA = "1st period margin: **" + printValue + "**."
        paramDict[firstPeriodMargin] = "1st period: **" + printValue + "**."
        if abs(marginVarianceSum) > 100:
            printValue = str(
                "{:,.1f}".format(round(periodZeroPercentMarginAfterCOGS, 1))
            )
        else:
            printValue = str(
                "{:,.1f}".format(round(periodZeroPercentMarginAfterCOGS, 1))
            )
        statusMessage1stB = (
            " 1st period margin as % of revenues: **" + printValue + "%**."
        )
        if abs(marginVarianceSum) > 100:
            printValue = str("{:,.0f}".format(round(periodOneMarginSum, 0)))
        else:
            printValue = str("{:,.1f}".format(round(periodOneMarginSum, 1)))
        statusMessage2ndA = " 2nd period margin: **" + printValue + "**. "
        paramDict[secondPeriodMargin] = " 2nd period: **" + printValue + "**. "
        if abs(marginVarianceSum) > 100:
            printValue = str(
                "{:,.1f}".format(round(periodOnePercentMarginAfterCOGS, 1))
            )
        else:
            printValue = str(
                "{:,.1f}".format(round(periodOnePercentMarginAfterCOGS, 1))
            )
        statusMessage2ndB = (
            " 2nd period margin as % of revenues: **" + printValue + "%**. "
        )
        paramDict = add_status_message_to_paramDict(
            paramDict, statusMessage1stA + " " + statusMessage1stB, 8
        )
        paramDict = add_status_message_to_paramDict(
            paramDict, statusMessage2ndA + " " + statusMessage2ndB, 8
        )
    return paramDict


def calculate_aggregated_indirect_cost_variance_value(
    df, paramDict, periodsArray, columns
):
    """
    if a discount columns exist, calculates the total variance after discount
    """
    namingParams = get_naming_params()
    separatorString = namingParams["separatorString"]
    varianceName = namingParams["varianceName"]
    totalAmountPeriodZero = paramDict[namingParams["totalAmountPeriodZero"]]
    totalAmountPeriodOne = paramDict[namingParams["totalAmountPeriodOne"]]
    indirectCostsColFound = namingParams["indirectCostsColFound"]
    netMarginName = namingParams["netMarginName"]
    netMarginVariance = namingParams["netMarginVariance"]
    indirectCostsVariance = namingParams["indirectCostsVariance"]
    percentVarianceAfterIndCosts = namingParams["percentVarianceAfterIndCosts"]
    indirectCostsName = namingParams["indirectCostsName"]
    totalNetMarginPeriodZeroKey = namingParams["totalNetMarginPeriodZero"]
    totalNetMarginPeriodOneKey = namingParams["totalNetMarginPeriodOne"]
    totalNetMarginPeriodZeroinPercent = namingParams[
        "totalNetMarginPeriodZeroinPercent"
    ]
    totalNetMarginPeriodOneinPercent = namingParams["totalNetMarginPeriodOneinPercent"]
    if indirectCostsColFound in paramDict and paramDict[indirectCostsColFound]:
        periodOnePercentMarginAfterIndCosts, periodZeroPercentMarginAfterIndCosts = 0, 0
        netMarginZero = netMarginName + separatorString + periodsArray[0]
        netMarginOne = netMarginName + separatorString + periodsArray[1]
        statusMessage = "No net margin value in dataframe"
        paramDict = add_status_message_to_paramDict(paramDict, statusMessage, 3)
        indirectCostsZero = indirectCostsName + separatorString + periodsArray[0]
        indirectCostsOne = indirectCostsName + separatorString + periodsArray[1]
        periodZeroNetMarginSum = df[netMarginZero].sum()
        periodOneNetMarginSum = df[netMarginOne].sum()
        periodZeroIndirectCostsSum = df[indirectCostsZero].sum()
        periodOneIndirectCostsSum = df[indirectCostsOne].sum()
        netMarginVarianceSum = periodOneNetMarginSum - periodZeroNetMarginSum
        indirectCostsVarianceSum = (
            periodOneIndirectCostsSum - periodZeroIndirectCostsSum
        )
        paramDict[indirectCostsVariance] = indirectCostsVarianceSum
        if totalAmountPeriodZero != 0:
            periodZeroPercentMarginAfterIndCosts = (
                periodZeroNetMarginSum / totalAmountPeriodZero * 100
            )
            paramDict[totalNetMarginPeriodZeroinPercent] = (
                periodZeroPercentMarginAfterIndCosts
            )
        if totalAmountPeriodOne != 0:
            periodOnePercentMarginAfterIndCosts = (
                periodOneNetMarginSum / totalAmountPeriodOne * 100
            )
            paramDict[totalNetMarginPeriodOneinPercent] = (
                periodOnePercentMarginAfterIndCosts
            )
        percentMarginAfterIndCostsChange = (
            periodOnePercentMarginAfterIndCosts - periodZeroPercentMarginAfterIndCosts
        )
        paramDict[percentVarianceAfterIndCosts] = percentMarginAfterIndCostsChange
        paramDict[netMarginVariance] = netMarginVarianceSum
        (
            paramDict[totalNetMarginPeriodZeroKey],
            paramDict[totalNetMarginPeriodOneKey],
        ) = (periodZeroNetMarginSum, periodOneNetMarginSum)
        if abs(netMarginVarianceSum) > 100:
            netMarginVarianceValue = str(
                "{:,.0f}".format(round(netMarginVarianceSum, 0))
            )
        else:
            netMarginVarianceValue = str(
                "{:,.1f}".format(round(netMarginVarianceSum, 1))
            )
        if periodZeroNetMarginSum != 0:
            yoyPercentChangeDelta = (
                periodOneNetMarginSum * 100 / periodZeroNetMarginSum
            ) - 100
            if abs(netMarginVarianceSum) > 100:
                yoyPercentChangeValue = str(
                    "{:,.1f}".format(round(yoyPercentChangeDelta, 1))
                )
            else:
                yoyPercentChangeValue = str(
                    "{:,.1f}".format(round(yoyPercentChangeDelta, 1))
                )
        else:
            yoyPercentChangeValue = nothingThereString

        statusMessage = [
            netMarginVariance + " " + varianceName,
            netMarginVarianceValue,
            yoyPercentChangeValue,
        ]
        statusMessage = check_if_reverse_KPI_color(
            paramDict,
            statusMessage,
            netMarginName,
            periodZeroNetMarginSum,
            periodOneNetMarginSum,
        )
        paramDict = add_status_message_to_paramDict(paramDict, statusMessage, 3)
        if abs(netMarginVarianceSum) > 100:
            printValue = str("{:,.0f}".format(round(periodZeroNetMarginSum, 0)))
        else:
            printValue = str("{:,.1f}".format(round(periodZeroNetMarginSum, 1)))
        statusMessage1stA = "1st period after indirect costs: **" + printValue + "**. "
        if abs(netMarginVarianceSum) > 100:
            printValue = str(
                "{:,.1f}".format(round(periodZeroPercentMarginAfterIndCosts, 1))
            )
        else:
            printValue = str(
                "{:,.1f}".format(round(periodZeroPercentMarginAfterIndCosts, 1))
            )
        statusMessage1stB = (
            "1st period after indirect costs as % of revenues: **"
            + printValue
            + "%**. "
        )
        if abs(netMarginVarianceSum) > 100:
            printValue = str("{:,.0f}".format(round(periodOneNetMarginSum, 0)))
        else:
            printValue = str("{:,.1f}".format(round(periodOneNetMarginSum, 1)))
        statusMessage2ndA = "2nd period after indirect costs: **" + printValue + "**. "
        if abs(netMarginVarianceSum) > 100:
            printValue = str(
                "{:,.1f}".format(round(periodOnePercentMarginAfterIndCosts, 1))
            )
        else:
            printValue = str(
                "{:,.1f}".format(round(periodOnePercentMarginAfterIndCosts, 1))
            )
        statusMessage2ndB = (
            "2nd period after indirect costs as % of revenues: **"
            + printValue
            + "%**. "
        )
        paramDict = add_status_message_to_paramDict(
            paramDict, statusMessage1stA + " " + statusMessage1stB, 10
        )
        paramDict = add_status_message_to_paramDict(
            paramDict, statusMessage2ndA + " " + statusMessage2ndB, 10
        )
    return paramDict


def calculate_total_amounts(df, paramDict, chartDict):
    """
    we write the total variance amount to a variable, as well as the period totals
    """
    namingParams = get_naming_params()
    configParams = get_config_params()
    periodsArray = configParams["periodsArray"]
    columns, schema = get_schema_and_column_names(df)
    paramDict = calculate_aggregated_price_volume_variance_value(
        df, paramDict, periodsArray, columns, chartDict
    )
    paramDict = calculate_aggregated_discount_variance_value(
        df, paramDict, periodsArray, columns
    )
    paramDict = calculate_aggregated_cogs_variance_value(
        df, paramDict, periodsArray, columns
    )
    paramDict = calculate_aggregated_indirect_cost_variance_value(
        df, paramDict, periodsArray, columns
    )
    return paramDict
