← Files ClaraARCHIVED FILE
modules/set-overlap-analysis/vendor/modules/data/identify_columns.py
108 KB · Oct 2, 2026 · 00:29 UTC
from __future__ import annotations
import copy
import datetime as dt
import logging
import sys
from pathlib import Path
import polars as pl
from dateutil.relativedelta import relativedelta
if "modules.utilities.utils" in sys.modules:
utils = sys.modules["modules.utilities.utils"]
else: # pragma: no cover
try:
import modules.utilities.utils as utils
except ModuleNotFoundError: # pragma: no cover
sys.path.append(str(Path(__file__).resolve().parents[2]))
import modules.utilities.utils as utils
from modules.charting.chart_primitives import check_if_plan_or_py
from modules.layout.memoization import check_collect
from modules.layout.session_manager import SessionManager
from modules.utilities.config import (
get_config_params,
get_naming_params,
get_run_params,
)
from modules.utilities.date_utils import parse_date_column
from modules.utilities.error_messages import (
add_app_message_to_paramdict,
add_error_message_in_load_data_tab,
add_info_message_in_period_options_tab,
add_warning_message_in_load_data_tab,
add_warning_message_in_period_options_tab,
add_write_message_in_period_options_tab,
)
from modules.utilities.helpers import (
drop_columns,
duplicate_dataframe,
get_dataset_specific_parameter,
get_date_columns_from_schema,
get_period_length,
get_periods_array,
measure_time,
print_error_details,
unique,
)
from modules.utilities.ui_notifier import ui as notifier
from src.date_helpers import start_of_month, start_of_quarter
from src.period_aggregators import (
calculate_period_to_date_same_year,
calculate_period_to_date_year_ago,
calculate_rolling_period,
)
logger = logging.getLogger(__name__)
def build_initial_index_array(
df: pl.DataFrame | pl.LazyFrame, paramDict: dict
) -> tuple[pl.LazyFrame, list[str], list[str], dict]:
"""
we want to separate the buildup of the index array in a separate function
so we can use feed different indexes to the function that indexes the df
here the order of the columns in the index does not matter
we also set all indexcolumns to object to avoid fake numeric columns
"""
was_lazy = isinstance(df, pl.LazyFrame)
df = df.lazy() if not was_lazy else df
namingParams = get_naming_params()
configParams = get_config_params()
stemArray = namingParams["stemArray"]
periodName = namingParams["periodName"]
dataPreparation = namingParams["dataPreparationName"]
indexArrayBuilt = namingParams["indexArrayBuiltName"]
cleanedNewIndexCols = namingParams["cleanedNewIndexColsName"]
toDropStemArray = configParams[namingParams["toDropStemArray"]]
metricsColsDict = configParams[namingParams["metricsColsDict"]]
nonMetricNumericColumns = namingParams["nonMetricNumericColumns"]
columns, schema = utils.get_schema_and_column_names(df)
indexCols = []
valueCols = []
toDrop = []
columns, schema = utils.get_schema_and_column_names(df)
for column in columns:
if column not in metricsColsDict:
if schema.get(column) != pl.Utf8:
toDrop = check_if_stems_in_string(column, toDropStemArray, toDrop)
if paramDict[namingParams["allDimensionsString"]]:
if schema.get(column) == pl.Utf8 or column == periodName:
indexCols.append(column)
else:
indexCols.append(column)
elif metricsColsDict[column]:
valueCols.append(column)
else:
toDrop.append(column)
newIndexCols = []
for indexCol in indexCols:
if indexCol not in toDrop:
newIndexCols.append(indexCol)
df = drop_columns(df, toDrop)
columns, schema = utils.get_schema_and_column_names(df)
paramDict[nonMetricNumericColumns] = []
for column in columns:
if column not in valueCols + newIndexCols:
paramDict[nonMetricNumericColumns].append(column)
measure_time(dataPreparation, indexArrayBuilt, False)
df = df.with_columns(
[
pl.col(column).cast(pl.Utf8).alias(column)
for column in newIndexCols
if column in columns
]
)
measure_time(dataPreparation, cleanedNewIndexCols, False)
return df, newIndexCols, valueCols, paramDict
try:
from src.identify_columns_logic import cogs_col_found as logic_cogs_col_found
from src.identify_columns_logic import date_col_found as logic_date_col_found
from src.identify_columns_logic import (
discount_col_found as logic_discount_col_found,
)
from src.identify_columns_logic import (
indirect_costs_col_found as logic_indirect_costs_col_found,
)
from src.identify_columns_logic import (
monetary_col_found as logic_monetary_col_found,
)
from src.identify_columns_logic import period_col_found as logic_period_col_found
from src.identify_columns_logic import units_col_found as logic_units_col_found
from src.identify_columns_logic import volume_col_found as logic_volume_col_found
except Exception as e: # pragma: no cover - missing in test stubs
logging.exception(e)
notifier.error("Something went wrong.")
logic_cogs_col_found = lambda param: (param, [])
logic_date_col_found = lambda param: (param, [])
logic_discount_col_found = lambda param: (param, [])
logic_indirect_costs_col_found = lambda param: (param, [])
logic_monetary_col_found = lambda param: (param, [])
logic_period_col_found = lambda param: (param, [])
logic_units_col_found = lambda param: (param, [])
logic_volume_col_found = lambda param: (param, [])
def monetary_col_found(paramDict, ui: "IdentifyColumnsUI" | None = None):
"""Display messages about the monetary column."""
if ui is None:
from ui.identify_columns_ui import IdentifyColumnsUI
ui = IdentifyColumnsUI()
paramDict, messages = logic_monetary_col_found(paramDict)
ui.show_messages(messages)
return paramDict, messages
def volume_col_found(paramDict, ui: "IdentifyColumnsUI" | None = None):
"""Display messages about the volume column."""
if ui is None:
from ui.identify_columns_ui import IdentifyColumnsUI
ui = IdentifyColumnsUI()
paramDict, messages = logic_volume_col_found(paramDict)
ui.show_messages(messages)
return paramDict, messages
def units_col_found(paramDict, ui: "IdentifyColumnsUI" | None = None):
"""Display messages about the units column."""
if ui is None:
from ui.identify_columns_ui import IdentifyColumnsUI
ui = IdentifyColumnsUI()
paramDict, messages = logic_units_col_found(paramDict)
ui.show_messages(messages)
return paramDict, messages
def date_col_found(paramDict, ui: "IdentifyColumnsUI" | None = None):
"""Display messages about detected date column."""
if ui is None:
from ui.identify_columns_ui import IdentifyColumnsUI
ui = IdentifyColumnsUI()
paramDict, messages = logic_date_col_found(paramDict)
ui.show_messages(messages)
return paramDict, messages
def period_col_found(
df,
paramDict,
ui: "IdentifyColumnsUI" | None = None,
preserve_date_col: bool = False,
):
"""Display messages about detected period column.
Parameters
----------
df: pl.LazyFrame | pl.DataFrame
Dataset under analysis.
paramDict: dict
Parameter dictionary used across data preparation.
ui: IdentifyColumnsUI | None, optional
UI helper for rendering messages.
preserve_date_col: bool, optional
When ``True`` the original date column is kept in the dataframe.
"""
if ui is None:
from ui.identify_columns_ui import IdentifyColumnsUI
ui = IdentifyColumnsUI()
paramDict, messages = logic_period_col_found(paramDict)
ui.show_messages(messages)
naming = get_naming_params()
if (
naming["periodColFound"] in paramDict
and paramDict[naming["periodColFound"]]
and len(paramDict.get(naming["likelyPeriodCols"], [])) > 0
and paramDict.get(naming["numberOfPeriodsFound"], 0) >= 1
):
if not preserve_date_col:
df = drop_columns(df, [naming["dateName"]])
return df, paramDict, messages
def discount_col_found(paramDict, ui: "IdentifyColumnsUI" | None = None):
"""Display messages about discount column."""
if ui is None:
from ui.identify_columns_ui import IdentifyColumnsUI
ui = IdentifyColumnsUI()
paramDict, messages = logic_discount_col_found(paramDict)
ui.show_messages(messages)
return paramDict, messages
def cogs_col_found(paramDict, ui: "IdentifyColumnsUI" | None = None):
"""Display messages about the COGS column."""
if ui is None:
from ui.identify_columns_ui import IdentifyColumnsUI
ui = IdentifyColumnsUI()
paramDict, messages = logic_cogs_col_found(paramDict)
ui.show_messages(messages)
return paramDict, messages
def indirect_costs_col_found(paramDict, ui: "IdentifyColumnsUI" | None = None):
"""Display messages about the indirect costs column."""
if ui is None:
from ui.identify_columns_ui import IdentifyColumnsUI
ui = IdentifyColumnsUI()
paramDict, messages = logic_indirect_costs_col_found(paramDict)
ui.show_messages(messages)
return paramDict, messages
def category_weighted_distribution_col_found(paramDict):
"""
messages to cwd monetary columns
"""
namingParams = get_naming_params()
configParams = get_config_params()
categoryWeightedDistributionName = namingParams["categoryWeightedDistributionName"]
cwdColFound = namingParams["cwdColFound"]
likelyCwdCols = namingParams["likelyCwdCols"]
successIcon = namingParams["successIcon"]
categoryWeightedDistributionStemDict = namingParams[
"categoryWeightedDistributionStemDict"
]
stemArray = namingParams["stemArray"]
stemDict = namingParams["stemDict"]
stemDict = configParams[stemDict]
if cwdColFound in paramDict:
nameList = (
str(stemDict[categoryWeightedDistributionStemDict][stemArray])
.replace("[", "")
.replace("]", "")
)
if paramDict[cwdColFound] and len(paramDict[likelyCwdCols]) > 0:
notifier.success(
"The **"
+ str(paramDict[likelyCwdCols][0])
+ "** column was tagged as the *category weighted distribution* driver column of the dataset.",
icon=successIcon,
)
notifier.caption(
"""If this is not correct, rename the category weighted distribution column and reload the file.
The category weighted distribution column must be in **plain number format** (no thousand separators, no % signs, zero written as 0 not as -).
The name of the category weighted distribution column **must contain** one of these stems:
"""
)
notifier.caption("*" + nameList + "*")
notifier.caption(""" \n""")
notifier.caption(""" \n""")
else:
notifier.caption(
"""We were unable to identify a *category weighted distribution* driver column in the dataset.
This is OK. The app **does not require** a category weighted distribution column.
"""
)
notifier.caption(
"""If your dataset indeed has a category weighted distribution column, rename it and reload the file.
The category weighted distribution column must be in **absolute (not percent)** value with positive sign.
The name of the category weighted distribution column **must contain** one of these stems:
"""
)
notifier.caption("*" + nameList + "*")
notifier.caption(""" \n""")
notifier.caption(""" \n""")
return paramDict
def checkouts_col_found(paramDict):
"""
messages to identify driver col
"""
namingParams = get_naming_params()
configParams = get_config_params()
checkoutsName = namingParams["checkoutsName"]
checkoutsColFound = namingParams["checkoutsColFound"]
likelyCheckoutsCols = namingParams["likelyCheckoutsCols"]
checkoutsStemDict = namingParams["checkoutsStemDict"]
successIcon = namingParams["successIcon"]
stemArray = namingParams["stemArray"]
stemDict = namingParams["stemDict"]
stemDict = configParams[stemDict]
if checkoutsColFound in paramDict:
nameList = (
str(stemDict[checkoutsStemDict][stemArray])
.replace("[", "")
.replace("]", "")
)
if paramDict[checkoutsColFound] and len(paramDict[likelyCheckoutsCols]) > 0:
notifier.success(
"The **"
+ str(paramDict[likelyCheckoutsCols][0])
+ "** was tagged as the *checkouts* driver column of the dataset.",
icon=successIcon,
)
notifier.caption(
"""If this is not correct, rename the checkouts column and reload the file.
The checkouts column must be in **plain number format** (no thousand separators, zero written as 0 not as -).
The name of the checkouts column **must contain** one of these stems:
"""
)
notifier.caption("*" + nameList + "*")
notifier.caption(""" \n""")
notifier.caption(""" \n""")
else:
notifier.caption(
"""We were unable to identify a *checkouts* driver column in the dataset.
This is OK. The app **does not require** a check out column.
"""
)
notifier.caption(
"""If your dataset indeed has a checkouts column, rename it and reload the file.
The checkouts column must be in **absolute (not percent)** value with positive sign.
The name of the checkouts column **must contain** one of these stems:
"""
)
notifier.caption("*" + nameList + "*")
notifier.caption(""" \n""")
notifier.caption(""" \n""")
return paramDict
def visits_col_found(paramDict):
"""
messages to identify driver col
"""
namingParams = get_naming_params()
configParams = get_config_params()
visitsName = namingParams["visitsName"]
visitsColFound = namingParams["visitsColFound"]
likelyVisitsCols = namingParams["likelyVisitsCols"]
visitsStemDict = namingParams["visitsStemDict"]
successIcon = namingParams["successIcon"]
stemArray = namingParams["stemArray"]
stemDict = namingParams["stemDict"]
stemDict = configParams[stemDict]
if visitsColFound in paramDict:
nameList = (
str(stemDict[visitsStemDict][stemArray]).replace("[", "").replace("]", "")
)
if paramDict[visitsColFound] and len(paramDict[likelyVisitsCols]) > 0:
notifier.success(
"The **"
+ str(paramDict[likelyVisitsCols][0])
+ "** was tagged as the *visits* driver column of the dataset.",
icon=successIcon,
)
notifier.caption(
"""If this is not correct, rename the visits column and reload the file.
The visits column must be in **plain number format** (no thousand separators, zero written as 0 not as -).
The name of the visits column **must contain** one of these stems:
"""
)
notifier.caption("*" + nameList + "*")
notifier.caption(""" \n""")
notifier.caption(""" \n""")
else:
notifier.caption(
"""We were unable to identify a *visits* drive column in the dataset.
This is OK. The app **does not require** a check out column.
"""
)
notifier.caption(
"""If your dataset indeed has a visits column, rename it and reload the file.
The visits column must be in **absolute (not percent)** value with positive sign.
The name of the visits column **must contain** one of these stems:
"""
)
notifier.caption("*" + nameList + "*")
notifier.caption(""" \n""")
notifier.caption(""" \n""")
return paramDict
def promo_amount_col_found(paramDict):
"""
messages to identify discount columns
"""
namingParams = get_naming_params()
configParams = get_config_params()
monetaryPromoLocalCurrencyName = namingParams["monetaryPromoLocalCurrencyName"]
monetaryPromoLocalCurrencyColFound = namingParams[
"monetaryPromoLocalCurrencyColFound"
]
likelyPromoLocalCurrencyValueCols = namingParams[
"likelyPromoLocalCurrencyValueCols"
]
monetaryPromoLocalCurrencyStemDict = namingParams[
"monetaryPromoLocalCurrencyStemDict"
]
successIcon = namingParams["successIcon"]
stemArray = namingParams["stemArray"]
stemDict = namingParams["stemDict"]
stemDict = configParams[stemDict]
if monetaryPromoLocalCurrencyColFound in paramDict:
nameList = (
str(stemDict[monetaryPromoLocalCurrencyStemDict][stemArray])
.replace("[", "")
.replace("]", "")
)
if (
paramDict[monetaryPromoLocalCurrencyColFound]
and len(paramDict[likelyPromoLocalCurrencyValueCols]) > 0
):
notifier.success(
"The **"
+ str(paramDict[likelyPromoLocalCurrencyValueCols][0])
+ "** column was tagged as the *promo amount* column of the dataset.",
icon=successIcon,
)
notifier.caption(
"""If this is not correct, rename the promo amount column and reload the file.
The promo amount column must be in **plain number format** (no thousand separators, no currency 💲 format, zero written as 0 not as -).
The name of the promo amount column **must contain** one of these stems:
"""
)
notifier.caption("*" + nameList + "*")
notifier.caption(""" \n""")
notifier.caption(""" \n""")
else:
notifier.caption(
"""We were unable to identify a *promo amount* column in the dataset.
This is OK. The app **does not require** a promo amount column.
"""
)
notifier.caption(
"""If your dataset indeed has a promo amount column, rename it and reload the file.
The name of the promo amount column **must contain** one of these stems:
"""
)
notifier.caption("*" + nameList + "*")
notifier.caption(""" \n""")
notifier.caption(""" \n""")
return paramDict
def no_promo_amount_col_found(paramDict):
"""
messages to identify discount columns
"""
namingParams = get_naming_params()
configParams = get_config_params()
monetaryNoPromoLocalCurrencyName = namingParams["monetaryNoPromoLocalCurrencyName"]
monetaryNoPromoLocalCurrencyColFound = namingParams[
"monetaryNoPromoLocalCurrencyColFound"
]
likelyNoPromoLocalCurrencyValueCols = namingParams[
"likelyNoPromoLocalCurrencyValueCols"
]
monetaryNoPromoLocalCurrencyStemDict = namingParams[
"monetaryNoPromoLocalCurrencyStemDict"
]
successIcon = namingParams["successIcon"]
stemArray = namingParams["stemArray"]
stemDict = namingParams["stemDict"]
stemDict = configParams[stemDict]
if monetaryNoPromoLocalCurrencyColFound in paramDict:
nameList = (
str(stemDict[monetaryNoPromoLocalCurrencyStemDict][stemArray])
.replace("[", "")
.replace("]", "")
)
if (
paramDict[monetaryNoPromoLocalCurrencyColFound]
and len(paramDict[likelyNoPromoLocalCurrencyValueCols]) > 0
):
notifier.success(
"The **"
+ str(paramDict[likelyNoPromoLocalCurrencyValueCols][0])
+ "** column was tagged as the *no promo amount* column of the dataset.",
icon=successIcon,
)
notifier.caption(
"""If this is not correct, rename the no promo amount column and reload the file.
The no promo amount column must be in **plain number format** (no thousand separators, no currency 💲 format, zero written as 0 not as -).
The name of the no promo amount column **must contain** one of these stems:
"""
)
notifier.caption("*" + nameList + "*")
notifier.caption(""" \n""")
notifier.caption(""" \n""")
else:
notifier.caption(
"""We were unable to identify a *no promo amount* column in the dataset.
This is OK. The app **does not require** a no promo amount column.
"""
)
notifier.caption(
"""If your dataset indeed has a no promo amount column, rename it and reload the file.
The name of the no promo amount column **must contain** one of these stems:
"""
)
notifier.caption("*" + nameList + "*")
notifier.caption(""" \n""")
notifier.caption(""" \n""")
return paramDict
def promo_units_col_found(paramDict):
"""
messages to identify discount columns
"""
namingParams = get_naming_params()
configParams = get_config_params()
unitsPromoName = namingParams["unitsPromoName"]
unitPromoColFound = namingParams["unitPromoColFound"]
likelyPromoUnitsCols = namingParams["likelyPromoUnitsCols"]
unitsPromoStemDict = namingParams["unitsPromoStemDict"]
successIcon = namingParams["successIcon"]
stemArray = namingParams["stemArray"]
stemDict = namingParams["stemDict"]
stemDict = configParams[stemDict]
if unitPromoColFound in paramDict:
nameList = (
str(stemDict[unitsPromoStemDict][stemArray])
.replace("[", "")
.replace("]", "")
)
if paramDict[unitPromoColFound] and len(paramDict[likelyPromoUnitsCols]) > 0:
notifier.success(
"The **"
+ str(paramDict[likelyPromoUnitsCols][0])
+ "** column was tagged as the *promo units* column of the dataset.",
icon=successIcon,
)
notifier.caption(
"""If this is not correct, rename the promo units column and reload the file.
The promo units column must be in **plain number format** (no thousand separators, no currency 💲 format, zero written as 0 not as -).
The name of the promo units column **must contain** one of these stems:
"""
)
notifier.caption("*" + nameList + "*")
notifier.caption(""" \n""")
notifier.caption(""" \n""")
else:
notifier.caption(
"""We were unable to identify a *promo units* column in the dataset.
This is OK. The app **does not require** a promo units column.
"""
)
notifier.caption(
"""If your dataset indeed has a promo units column, rename it and reload the file.
The name of the promo units column **must contain** one of these stems:
"""
)
notifier.caption("*" + nameList + "*")
notifier.caption(""" \n""")
notifier.caption(""" \n""")
return paramDict
def no_promo_units_col_found(paramDict):
"""
messages to identify discount columns
"""
namingParams = get_naming_params()
configParams = get_config_params()
unitsNoPromoName = namingParams["unitsNoPromoName"]
unitNoPromoColFound = namingParams["unitNoPromoColFound"]
likelyNoPromoUnitsCols = namingParams["likelyNoPromoUnitsCols"]
unitsNoPromoStemDict = namingParams["unitsNoPromoStemDict"]
successIcon = namingParams["successIcon"]
stemArray = namingParams["stemArray"]
stemDict = namingParams["stemDict"]
stemDict = configParams[stemDict]
if unitNoPromoColFound in paramDict:
nameList = (
str(stemDict[unitsNoPromoStemDict][stemArray])
.replace("[", "")
.replace("]", "")
)
if (
paramDict[unitNoPromoColFound]
and len(paramDict[likelyNoPromoUnitsCols]) > 0
):
notifier.success(
"The **"
+ str(paramDict[likelyNoPromoUnitsCols][0])
+ "** column was tagged as the *no promo units* column of the dataset.",
icon=successIcon,
)
notifier.caption(
"""If this is not correct, rename the no promo units column and reload the file.
The no promo units column must be in **plain number format** (no thousand separators, no currency 💲 format, zero written as 0 not as -).
The name of the no promo units column **must contain** one of these stems:
"""
)
notifier.caption("*" + nameList + "*")
notifier.caption(""" \n""")
notifier.caption(""" \n""")
else:
notifier.caption(
"""We were unable to identify a *no promo units* column in the dataset.
This is OK. The app **does not require** a no promo units column.
"""
)
notifier.caption(
"""If your dataset indeed has a no promo units column, rename it and reload the file.
The name of the no promo units column **must contain** one of these stems:
"""
)
notifier.caption("*" + nameList + "*")
notifier.caption(""" \n""")
notifier.caption(""" \n""")
return paramDict
from src.identify_columns_logic import show_input_data as logic_show_input_data
def show_input_data(
df: pl.LazyFrame, paramDict: dict
) -> tuple[pl.LazyFrame, dict, list[tuple[str, str, str | None]]]:
"""Return messages about detected columns without rendering UI."""
namingParams = get_naming_params()
impossibleToProcessFile = namingParams["impossibleToProcessFile"]
stemDict = namingParams["stemDict"]
messages: list[tuple[str, str, str | None]] = []
if utils.is_valid_lazyframe(df):
df, paramDict, messages = logic_show_input_data(df, paramDict)
if impossibleToProcessFile in paramDict and paramDict[impossibleToProcessFile]:
df = pl.LazyFrame()
paramDict.pop(stemDict, None)
return df, paramDict, messages
def check_for_datecolumns_in_schema(df):
columns, schema = utils.get_schema_and_column_names(df)
dateColumns = get_date_columns_from_schema(schema)
return dateColumns, columns
def find_and_parse_datecolumns(
df, paramDict, session_manager: SessionManager | None = None
):
configParams = get_config_params()
namingParams = get_naming_params()
dateStemDict = namingParams["dateStemDict"]
stemDict = namingParams["stemDict"]
dateColFound = namingParams["dateColFound"]
likelyDateColsKey = namingParams["likelyDateCols"]
notMetConditionValue = namingParams["notMetConditionValue"]
metConditionValue = namingParams["metConditionValue"]
impossibleToProcessFile = namingParams["impossibleToProcessFile"]
stemDict = configParams[stemDict]
stemDict = stemDict[dateStemDict]
stemArray = stemDict[namingParams["stemArray"]]
notStemArray = stemDict[namingParams["notStemArray"]]
possibleDateCols = []
likelyDateCols = []
paramDict[dateColFound] = notMetConditionValue
dateColumns, columns = check_for_datecolumns_in_schema(df)
session_manager = session_manager or SessionManager()
if len(dateColumns) == 0:
for column in columns:
possibleDateCols, toDrop = check_if_col_name_in_array(
column, stemArray, notStemArray, possibleDateCols, []
)
if len(possibleDateCols) > 0:
for dateCol in possibleDateCols:
try:
df = parse_date_column(
df,
dateCol,
session_manager=session_manager,
drop_invalid=False,
)
likelyDateCols.append(dateCol)
except Exception as e: # pragma: no cover - rare date formats
logging.exception(e)
msg = print_error_details(e)
paramDict = add_error_message_in_load_data_tab(paramDict, msg)
paramDict[impossibleToProcessFile] = True
paramDict["dateParseError"] = msg
paramDict[likelyDateColsKey] = likelyDateCols
if len(likelyDateCols) > 0:
paramDict[dateColFound] = metConditionValue
else:
paramDict[likelyDateColsKey] = dateColumns
paramDict[dateColFound] = metConditionValue
return df, paramDict
def check_if_stems_in_string(string, checkArray, resultArray):
"""
we check if a given string contains a stem that belongs to an array
"""
for stem in checkArray:
if stem.lower() in string.lower():
if string not in resultArray:
resultArray.append(string)
return resultArray
def check_if_col_name_in_array(
column, stemArray, notStemArray, possibleColsArray, toDrop
):
"""
checking if column name matches some stem that identifies column type
"""
possibleColsArray = check_if_stems_in_string(column, stemArray, possibleColsArray)
toDrop = check_if_stems_in_string(column, notStemArray, toDrop)
return possibleColsArray, toDrop
def fill_na_columns(df, column, fillValue):
"""Fill null values in ``column`` with ``fillValue``."""
return df.with_columns(pl.col(column).fill_null(fillValue))
def detect_column_name(
df, metricName, stemsDict, paramDict, checkIfNumeric, checkIfDate, toFloat
):
"""
we try to find the columns that contain the dollar, the value metrics
we have two array. The first of stems that should match the second of stems that should not
match, for value and volume metrics
We also delete columns such as unit price that we do not want
"""
runParams = get_run_params()
checkDateColName = runParams["checkDateColName"]
namingParams = get_naming_params()
possibleColsArray = []
noColsArray = []
likelyColsArray = []
toDrop = []
dateName = namingParams["dateName"]
periodName = namingParams["periodName"]
periodColFound = namingParams["periodColFound"]
uploadedFileType = namingParams["uploadedFileType"]
nothingThereString = namingParams["nothingThereString"]
errorMessageType = namingParams["errorMessageType"]
loadDataTabKey = namingParams["loadDataTab"]
impossibleToProcessFile = namingParams["impossibleToProcessFile"]
stemArray = stemsDict[namingParams["stemArray"]]
notStemArray = stemsDict[namingParams["notStemArray"]]
getIndex = stemsDict[namingParams["getIndex"]]
numericTypesArray = [
pl.Int8,
pl.Int16,
pl.Int32,
pl.Int64,
pl.UInt8,
pl.UInt16,
pl.UInt32,
pl.UInt64,
pl.Float32,
pl.Float64,
]
columns, schema = utils.get_schema_and_column_names(df)
colNumber = 0
schemaDict = dict(schema) if schema else {}
for column in columns:
possibleColsArray, toDrop = check_if_col_name_in_array(
column, stemArray, notStemArray, possibleColsArray, toDrop
)
for possibleCol in possibleColsArray:
for notStem in notStemArray:
if notStem in possibleCol:
noColsArray.append(possibleCol)
for possibleCol in possibleColsArray:
if possibleCol not in noColsArray:
likelyColsArray.append(possibleCol)
likelyColsArray = list(set(likelyColsArray))
isFound = True
found_cols = paramDict.get(namingParams["foundColsArray"], [])
# Case 1: the metric column already exists and hasn’t been used
if metricName in columns and metricName not in found_cols:
paramDict[namingParams["foundColsArray"]].append(metricName)
if toFloat:
df = fill_na_columns(df, metricName, 0)
try:
df = df.with_columns(
pl.col(metricName).cast(pl.Float64).alias(metricName)
)
except Exception as e:
logging.exception(e)
e = print_error_details(e)
paramDict = add_app_message_to_paramdict(
e,
errorMessageType,
loadDataTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=colNumber,
)
message = "Unable to parse numeric column. Make sure it is in plain number format"
paramDict = add_app_message_to_paramdict(
message,
errorMessageType,
loadDataTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=colNumber,
)
paramDict[impossibleToProcessFile] = True
elif (
len(likelyColsArray) > getIndex and likelyColsArray[getIndex] not in found_cols
):
candidate_col = likelyColsArray[getIndex]
if schema.get(candidate_col) in numericTypesArray:
df = df.rename({candidate_col: metricName}).with_columns(
pl.col(metricName).fill_null(0).alias(metricName)
)
else:
df = df.rename({candidate_col: metricName})
paramDict[namingParams["foundColsArray"]].append(candidate_col)
if toFloat:
try:
df = df.with_columns(
pl.col(metricName).cast(pl.Float64).alias(metricName)
)
except Exception as e:
logging.exception(e)
e = print_error_details(e)
paramDict = add_app_message_to_paramdict(
e,
errorMessageType,
loadDataTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=colNumber,
)
message = "Unable to parse numeric column. Make sure it is in plain number format"
paramDict = add_app_message_to_paramdict(
message,
errorMessageType,
loadDataTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=colNumber,
)
paramDict[impossibleToProcessFile] = True
else:
isFound = False
if len(likelyColsArray) == 0:
isFound = False
return df, isFound, likelyColsArray, paramDict
def detect_column(
df,
paramDict,
columnName,
columnStemDict,
colFound,
likelyCols,
checkIfNumeric,
checkIfDate,
toFloat,
):
"""
putting functions together
"""
namingParams = get_naming_params()
columnStems, paramDict = get_dataset_specific_parameter(
paramDict, namingParams["stemDict"], columnStemDict
)
df, isColFound, likelyColsArray, paramDict = detect_column_name(
df, columnName, columnStems, paramDict, checkIfNumeric, checkIfDate, toFloat
)
paramDict[colFound], paramDict[likelyCols] = isColFound, likelyColsArray
return paramDict, df
def insert_calculated_difference_column(df, fromCol, subtractCol, resultCol):
"""
we recalculate, and insert if they are not already there, calculated columns such as net sales or margin after cogs
so we do not have to worry if they are missing or wrongly calculated
"""
columns, schema = utils.get_schema_and_column_names(df)
if subtractCol in columns:
if fromCol in columns:
df = df.with_columns(
(pl.col(fromCol) - pl.col(subtractCol)).alias(resultCol)
)
return df
def find_distribution_metric_columns(df, paramDict):
"""
we try to find the columns that contain the dollar, the value and the date metrics
based on stem matching. First we check if we already have the right columns
"""
namingParams = get_naming_params()
categoryWeightedDistributionName = namingParams["categoryWeightedDistributionName"]
totalDistributionPointsName = namingParams["totalDistributionPointsName"]
baselineValueMetricName = namingParams["baselineValueMetricName"]
baselineUnitsMetricName = namingParams["baselineUnitsMetricName"]
likelyCwdCols = namingParams["likelyCwdCols"]
likelyTdpCols = namingParams["likelyTdpCols"]
likelyBaselineValueCols = namingParams["likelyBaselineValueCols"]
likelyBaselineUnitsCols = namingParams["likelyBaselineUnitsCols"]
cwdColFound = namingParams["cwdColFound"]
tdpColFound = namingParams["tdpColFound"]
baselineValueColFound = namingParams["baselineValueColFound"]
baselineUnitsColFound = namingParams["baselineUnitsColFound"]
categoryWeightedDistributionStemDict = namingParams[
"categoryWeightedDistributionStemDict"
]
totalDistributionPointsStemDict = namingParams["totalDistributionPointsStemDict"]
baselineValueStemDict = namingParams["baselineValueStemDict"]
baselineunitsStemDict = namingParams["baselineunitsStemDict"]
paramDict[namingParams["foundColsArray"]] = []
paramDict, df = detect_column(
df,
paramDict,
categoryWeightedDistributionName,
categoryWeightedDistributionStemDict,
cwdColFound,
likelyCwdCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
totalDistributionPointsName,
totalDistributionPointsStemDict,
tdpColFound,
likelyTdpCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
baselineValueMetricName,
baselineValueStemDict,
baselineValueColFound,
likelyBaselineValueCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
baselineUnitsMetricName,
baselineunitsStemDict,
baselineUnitsColFound,
likelyBaselineUnitsCols,
True,
False,
True,
)
return df, paramDict
def find_checkout_columns(df, paramDict):
"""
we try to find the columns that contain the dollar, the value and the date metrics
based on stem matching. First we check if we already have the right columns
"""
namingParams = get_naming_params()
checkoutsName = namingParams["checkoutsName"]
likelyCheckoutsCols = namingParams["likelyCheckoutsCols"]
checkoutsColFound = namingParams["checkoutsColFound"]
checkoutsStemDict = namingParams["checkoutsStemDict"]
paramDict[namingParams["foundColsArray"]] = []
paramDict, df = detect_column(
df,
paramDict,
checkoutsName,
checkoutsStemDict,
checkoutsColFound,
likelyCheckoutsCols,
True,
False,
True,
)
return df, paramDict
def find_visits_columns(df, paramDict):
"""
we try to find the columns that contain the dollar, the value and the date metrics
based on stem matching. First we check if we already have the right columns
"""
namingParams = get_naming_params()
visitsName = namingParams["visitsName"]
likelyVisitsCols = namingParams["likelyVisitsCols"]
visitsColFound = namingParams["visitsColFound"]
visitsStemDict = namingParams["visitsStemDict"]
paramDict[namingParams["foundColsArray"]] = []
paramDict, df = detect_column(
df,
paramDict,
visitsName,
visitsStemDict,
visitsColFound,
likelyVisitsCols,
True,
False,
True,
)
return df, paramDict
def find_volume_columns(df, paramDict):
"""
we try to find the columns that contain the dollar, the value and the date metrics
based on stem matching. First we check if we already have the right columns
"""
namingParams = get_naming_params()
unitsName = namingParams["unitsName"]
volumeName = namingParams["volumeName"]
unitsPromoName = namingParams["unitsPromoName"]
unitsNoPromoName = namingParams["unitsNoPromoName"]
likelyUnitsCols = namingParams["likelyUnitsCols"]
likelyVolumeCols = namingParams["likelyVolumeCols"]
likelyPromoUnitsCols = namingParams["likelyPromoUnitsCols"]
likelyNoPromoUnitsCols = namingParams["likelyNoPromoUnitsCols"]
unitsColFound = namingParams["unitsColFound"]
volumeColFound = namingParams["volumeColFound"]
unitPromoColFound = namingParams["unitPromoColFound"]
unitNoPromoColFound = namingParams["unitNoPromoColFound"]
unitsStemDict = namingParams["unitsStemDict"]
volumeStemDict = namingParams["volumeStemDict"]
unitsPromoStemDict = namingParams["unitsPromoStemDict"]
unitsNoPromoStemDict = namingParams["unitsNoPromoStemDict"]
paramDict[namingParams["foundColsArray"]] = []
paramDict, df = detect_column(
df,
paramDict,
unitsName,
unitsStemDict,
unitsColFound,
likelyUnitsCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
volumeName,
volumeStemDict,
volumeColFound,
likelyVolumeCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
unitsPromoName,
unitsPromoStemDict,
unitPromoColFound,
likelyPromoUnitsCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
unitsNoPromoName,
unitsNoPromoStemDict,
unitNoPromoColFound,
likelyNoPromoUnitsCols,
True,
False,
True,
)
return df, paramDict
def find_monetary_columns(df, paramDict):
"""
we try to find the columns that contain the dollar, the value and the date metrics
based on stem matching. First we check if we already have the right columns
"""
namingParams = get_naming_params()
monetaryName = namingParams["monetaryLocalCurrencyName"]
monetaryPromoLocalCurrencyName = namingParams["monetaryPromoLocalCurrencyName"]
monetaryNoPromoLocalCurrencyName = namingParams["monetaryNoPromoLocalCurrencyName"]
monetaryDollarCurrencyName = namingParams["monetaryDollarCurrencyName"]
monetaryPromoDollarCurrencyName = namingParams["monetaryPromoDollarCurrencyName"]
monetaryNoPromoDollarCurrencyName = namingParams[
"monetaryNoPromoDollarCurrencyName"
]
likelyValueCols = namingParams["likelyLocalCurrencyValueCols"]
likelyPromoLocalCurrencyValueCols = namingParams[
"likelyPromoLocalCurrencyValueCols"
]
likelyNoPromoLocalCurrencyValueCols = namingParams[
"likelyNoPromoLocalCurrencyValueCols"
]
likelyDollarCurrencyValueCols = namingParams["likelyDollarCurrencyValueCols"]
likelyPromoDollarCurrencyValueCols = namingParams[
"likelyPromoDollarCurrencyValueCols"
]
likelyNoPromoDollarCurrencyValueCols = namingParams[
"likelyNoPromoDollarCurrencyValueCols"
]
monetaryColFound = namingParams["monetaryLocalCurrencyColFound"]
monetaryPromoLocalCurrencyColFound = namingParams[
"monetaryPromoLocalCurrencyColFound"
]
monetaryNoPromoLocalCurrencyColFound = namingParams[
"monetaryNoPromoLocalCurrencyColFound"
]
monetaryDollarCurrencyColFound = namingParams["monetaryDollarCurrencyColFound"]
monetaryPromoDollarCurrencyColFound = namingParams[
"monetaryPromoDollarCurrencyColFound"
]
monetaryNoPromoDollarCurrencyColFound = namingParams[
"monetaryNoPromoDollarCurrencyColFound"
]
monetaryStemDict = namingParams["monetaryLocalCurrencyStemDict"]
monetaryPromoLocalCurrencyStemDict = namingParams[
"monetaryPromoLocalCurrencyStemDict"
]
monetaryNoPromoLocalCurrencyStemDict = namingParams[
"monetaryNoPromoLocalCurrencyStemDict"
]
monetaryDollarCurrencyStemDict = namingParams["monetaryDollarCurrencyStemDict"]
monetaryPromoDollarCurrencyStemDict = namingParams[
"monetaryPromoDollarCurrencyStemDict"
]
monetaryNoPromoDollarCurrencyStemDict = namingParams[
"monetaryNoPromoDollarCurrencyStemDict"
]
paramDict[namingParams["foundColsArray"]] = []
paramDict, df = detect_column(
df,
paramDict,
monetaryName,
monetaryStemDict,
monetaryColFound,
likelyValueCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
monetaryDollarCurrencyName,
monetaryDollarCurrencyStemDict,
monetaryDollarCurrencyColFound,
likelyDollarCurrencyValueCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
monetaryPromoDollarCurrencyName,
monetaryPromoDollarCurrencyStemDict,
monetaryPromoDollarCurrencyColFound,
likelyPromoDollarCurrencyValueCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
monetaryNoPromoDollarCurrencyName,
monetaryNoPromoDollarCurrencyStemDict,
monetaryNoPromoDollarCurrencyColFound,
likelyNoPromoDollarCurrencyValueCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
monetaryPromoLocalCurrencyName,
monetaryPromoLocalCurrencyStemDict,
monetaryPromoLocalCurrencyColFound,
likelyPromoLocalCurrencyValueCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
monetaryNoPromoLocalCurrencyName,
monetaryNoPromoLocalCurrencyStemDict,
monetaryNoPromoLocalCurrencyColFound,
likelyNoPromoLocalCurrencyValueCols,
True,
False,
True,
)
if paramDict[monetaryColFound] or paramDict.get(
namingParams["cogsColFound"], False
):
pass
else:
paramDict[namingParams["impossibleToProcessFile"]] = True
return df, paramDict
def find_discount_and_cogs_columns(df, paramDict):
"""
we try to find the columns that contain the dollar, the value and the date metrics
based on stem matching. First we check if we already have the right columns
"""
namingParams = get_naming_params()
monetaryName = namingParams["monetaryLocalCurrencyName"]
discountName = namingParams["discountName"]
netOfDiscountName = namingParams["netOfDiscountName"]
cogsName = namingParams["cogsName"]
marginName = namingParams["marginName"]
monetaryColFound = namingParams["monetaryLocalCurrencyColFound"]
impossibleToProcessFile = namingParams["impossibleToProcessFile"]
likelyDiscountCols = namingParams["likelyDiscountCols"]
likelyNetOfDiscountCols = namingParams["likelyNetOfDiscountCols"]
likelyCogsCols = namingParams["likelyCogsCols"]
likelyMarginCols = namingParams["likelyMarginCols"]
discountColFound = namingParams["discountColFound"]
netOfDiscountColFound = namingParams["netOfDiscountColFound"]
cogsColFound = namingParams["cogsColFound"]
marginColFound = namingParams["marginColFound"]
discountStemDict = namingParams["discountStemDict"]
netOfDiscountStemDict = namingParams["netOfDiscountStemDict"]
cogsStemDict = namingParams["cogsStemDict"]
marginStemDict = namingParams["marginStemDict"]
indirectCostsName = namingParams["indirectCostsName"]
indirectCostsStemDict = namingParams["indirectCostsStemDict"]
indirectCostsColFound = namingParams["indirectCostsColFound"]
likelyIndirectCostsCols = namingParams["likelyIndirectCostsCols"]
netMarginName = namingParams["netMarginName"]
netMarginStemDict = namingParams["netMarginStemDict"]
netMarginColFound = namingParams["netMarginColFound"]
likelyNetMarginCols = namingParams["likelyNetMarginCols"]
paramDict[namingParams["foundColsArray"]] = []
paramDict, df = detect_column(
df,
paramDict,
discountName,
discountStemDict,
discountColFound,
likelyDiscountCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
netOfDiscountName,
netOfDiscountStemDict,
netOfDiscountColFound,
likelyNetOfDiscountCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
indirectCostsName,
indirectCostsStemDict,
indirectCostsColFound,
likelyIndirectCostsCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
netMarginName,
netMarginStemDict,
netMarginColFound,
likelyNetMarginCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
cogsName,
cogsStemDict,
cogsColFound,
likelyCogsCols,
True,
False,
True,
)
paramDict, df = detect_column(
df,
paramDict,
marginName,
marginStemDict,
marginColFound,
likelyMarginCols,
True,
False,
True,
)
columns_status = [
(monetaryName, paramDict.get(monetaryColFound)),
(discountName, paramDict.get(discountColFound)),
(netOfDiscountName, paramDict.get(netOfDiscountColFound)),
(cogsName, paramDict.get(cogsColFound)),
(marginName, paramDict.get(marginColFound)),
]
found_cols = [name for name, found in columns_status if found]
missing_cols = [name for name, found in columns_status if not found]
logger.debug("Found monetary columns: %s", ", ".join(found_cols) or "none")
logger.debug("Missing monetary columns: %s", ", ".join(missing_cols) or "none")
columns, schema = utils.get_schema_and_column_names(df)
if cogsName in columns and discountName in columns:
df = insert_calculated_difference_column(
df, monetaryName, discountName, netOfDiscountName
)
df = insert_calculated_difference_column(
df, netOfDiscountName, cogsName, marginName
)
elif cogsName in columns:
df = insert_calculated_difference_column(df, monetaryName, cogsName, marginName)
elif discountName in columns:
df = insert_calculated_difference_column(
df, monetaryName, discountName, netOfDiscountName
)
if marginName in columns and indirectCostsName in columns:
df = insert_calculated_difference_column(
df, marginName, indirectCostsName, netMarginName
)
baseline_found = paramDict.get(monetaryColFound) or paramDict.get(cogsColFound)
paramDict[impossibleToProcessFile] = not baseline_found
return df, paramDict
def find_date_and_period_columns(df, paramDict):
"""
we try to find the columns that contain the dollar, the value and the date metrics
based on stem matching. First we check if we already have the right columns
"""
namingParams = get_naming_params()
dateName = namingParams["dateName"]
periodName = namingParams["periodName"]
likelyDateCols = namingParams["likelyDateCols"]
likelyPeriodCols = namingParams["likelyPeriodCols"]
periodName = namingParams["periodName"]
dateColFound = namingParams["dateColFound"]
periodColFound = namingParams["periodColFound"]
dateStemDict = namingParams["dateStemDict"]
periodStemDict = namingParams["periodStemDict"]
paramDict[namingParams["foundColsArray"]] = []
paramDict[namingParams["impossibleToProcessFile"]] = False
errorMessageType = namingParams["errorMessageType"]
loadDataTabKey = namingParams["loadDataTab"]
paramDict, df = detect_column(
df,
paramDict,
periodName,
periodStemDict,
periodColFound,
likelyPeriodCols,
False,
False,
False,
)
paramDict, df = detect_column(
df,
paramDict,
dateName,
dateStemDict,
dateColFound,
likelyDateCols,
False,
True,
False,
)
if paramDict[periodColFound]:
try:
df = df.with_columns(
pl.col(periodName).str.to_uppercase().alias(periodName)
)
except Exception as e: # noqa: BLE001
logging.exception(e)
e = print_error_details(e)
paramDict = add_app_message_to_paramdict(
e,
errorMessageType,
loadDataTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=0,
)
elif paramDict[dateColFound]:
pass
else:
paramDict[namingParams["impossibleToProcessFile"]] = True
return df, paramDict
def manage_trailing_minus(df: pl.LazyFrame, column: str) -> pl.LazyFrame:
# Check if the column is in the schema
columns, schema = utils.get_schema_and_column_names(df)
if column not in schema:
notifier.warning(f"Column '{column}' not found. Skipping.")
return df
# If the column is not a string, skip the fix
if schema[column] != pl.Utf8:
pass
return df
# If it's string, then apply your trailing-minus fix
try:
updated_df = df.with_columns(
pl.col(column)
.str.replace(r"^(.+)([-+])$", r"$2$1") # move trailing sign to front
.cast(pl.Float64)
.alias(column)
)
notifier.caption(
f"The **{column}** column was converted from trailing sign (e.g. `20-`) "
"to leading sign (e.g. `-20`) and cast to float."
)
return updated_df
except Exception as e:
logging.exception(e)
notifier.warning(f"Could not manage trailing minus for column `{column}`: {e}")
return df
def check_for_string_metric_cols(df, paramDict):
"""
this check if there are "numeric" columns with names such as "amount", "units", and so on
that are in string format for some reason (for instance trailing minus in SAP negative values)
and then it can modify the columns and make them numeric
"""
namingParams = get_naming_params()
configParams = get_config_params()
unitsStemDict = namingParams["unitsStemDict"]
volumeStemDict = namingParams["volumeStemDict"]
monetaryLocalCurrencyStemDict = namingParams["monetaryLocalCurrencyStemDict"]
discountStemDict = namingParams["discountStemDict"]
cogsStemDict = namingParams["cogsStemDict"]
stemArray = namingParams["stemArray"]
notStemArray = namingParams["notStemArray"]
stemDict = namingParams["stemDict"]
stemDict = configParams[stemDict]
columns, schema = utils.get_schema_and_column_names(df)
numericDictArray = [
unitsStemDict,
volumeStemDict,
monetaryLocalCurrencyStemDict,
discountStemDict,
cogsStemDict,
]
schemaDict = {}
schemaDict = dict(df.collect_schema())
for column in columns:
is_string = schemaDict[column]
if is_string:
for wordDict in numericDictArray:
if wordDict in stemDict:
if column in stemDict[wordDict][stemArray]:
if column not in stemDict[wordDict][notStemArray]:
df = manage_trailing_minus(df, column)
else:
pass
else:
pass
return df, paramDict
def delete_rows_with_all_zeroes(df, paramDict):
"""
no need to keep rows where both amount and quantity are zero
we do not do this for ROI calculation because otherwise the division is wrong
we also drop all the rows where the amount and the unit column have opposite sign
"""
namingParams = get_naming_params()
unitsName = namingParams["unitsName"]
volumeName = namingParams["volumeName"]
monetaryName = namingParams["monetaryLocalCurrencyName"]
unitsColFound = namingParams["unitsColFound"]
volumeColFound = namingParams["volumeColFound"]
metConditionValue = namingParams["metConditionValue"]
monetaryColFound = namingParams["monetaryLocalCurrencyColFound"]
isunitsColFound = paramDict[unitsColFound]
isvolumeColFound = paramDict[volumeColFound]
ismonetaryColFound = paramDict[monetaryColFound]
if isunitsColFound and ismonetaryColFound:
df = df.filter(
((pl.col(unitsName) != 0) | (pl.col(monetaryName) != 0))
& (
((pl.col(unitsName) >= 0) & (pl.col(monetaryName) >= 0))
| ((pl.col(unitsName) <= 0) & (pl.col(monetaryName) <= 0))
)
)
if isvolumeColFound and ismonetaryColFound:
df = df.filter(
((pl.col(volumeName) != 0) | (pl.col(monetaryName) != 0))
& (
((pl.col(volumeName) >= 0) & (pl.col(monetaryName) >= 0))
| ((pl.col(volumeName) <= 0) & (pl.col(monetaryName) <= 0))
)
)
if not utils.is_valid_lazyframe(df):
if isunitsColFound:
message = "The app automatically deletes the rows that have both amount and units column equal to zero or with opposite sign."
paramDict = add_warning_message_in_load_data_tab(paramDict, message)
if isvolumeColFound:
message = "The app automatically deletes the rows that have both amount and volume column equal to zero or with opposite sign."
paramDict = add_warning_message_in_load_data_tab(paramDict, message)
message = "All rows have both the amount or the units/volume column equal to zero, or with opposite sign"
paramDict = add_warning_message_in_load_data_tab(paramDict, message)
elif ismonetaryColFound:
df = df.filter(pl.col(monetaryName) != 0)
return df
def clean_PL_rows(dfCopy, paramDict, chartDict):
"""
if there is scenario data in the dataset, we take out the PL rows in order to show time line trend data
"""
namingParams = get_naming_params()
configParams = get_config_params()
planStemArray = configParams[namingParams["planStemArray"]]
timelineChart = namingParams["timelineChart"]
selectedPeriods = namingParams["selectedPeriods"]
areaChart = namingParams["areaChart"]
multitierColumnChart = namingParams["multitierColumnChart"]
horizontalWaterfallChart = namingParams["horizontalWaterfallChart"]
stackedColumnChart = namingParams["stackedColumnChart"]
trendComparisonByPeriodChart = namingParams["trendComparisonByPeriodChart"]
periodName = namingParams["periodName"]
scenarioName = namingParams["scenarioName"]
mostRecentDateKey = namingParams["mostRecentDate"]
pyName = namingParams["pyName"]
plName = namingParams["plName"]
fcName = namingParams["fcName"]
acName = namingParams["acName"]
dateName = namingParams["dateName"]
metConditionValue = namingParams["metConditionValue"]
comparePeriods = namingParams["comparePeriods"]
compareScenariosOrPeriods = namingParams["compareScenariosOrPeriods"]
compareWithYearBefore = namingParams["compareWithYearBefore"]
periodColFound = namingParams["periodColFound"]
filterDates = namingParams["filterDates"]
notMetConditionValue = namingParams["notMetConditionValue"]
chosenChart = namingParams["chosenChart"]
if chosenChart in chartDict:
chosenChart = chartDict[chosenChart]
keepFurePlanValues = notMetConditionValue
if (
compareScenariosOrPeriods in chartDict
and chartDict[compareScenariosOrPeriods] == comparePeriods
):
df = duplicate_dataframe(dfCopy)
if isinstance(df, pl.DataFrame):
df = df.lazy()
uniquePeriods = get_periods_array(df)
isExpectedData, planName = check_if_plan_or_py(uniquePeriods)
if plName in uniquePeriods and isExpectedData:
dfAC = df.filter(pl.col(periodName) == acName)
mostRecentACDate = dfAC.select(pl.col(dateName).max()).item()
dfPL = df.filter(pl.col(periodName) == plName)
mostRecentPLDate = dfPL.select(pl.col(dateName).max()).item()
if (
mostRecentPLDate > mostRecentACDate
): # and chosenChart in [stackedColumnChart]:
if (
compareWithYearBefore in chartDict
and not chartDict[compareWithYearBefore]
):
df = df.filter(
(pl.col(periodName).is_in([acName]))
| (pl.col(dateName) > mostRecentACDate)
).with_columns(pl.col(periodName).alias(scenarioName))
else:
df = df.filter(~pl.col(periodName).is_in([plName, fcName]))
else:
df = df.filter(~pl.col(periodName).is_in([plName, fcName]))
df = drop_columns(df, [periodName])
paramDict[periodColFound], chartDict[filterDates] = (
notMetConditionValue,
notMetConditionValue,
)
return df.lazy() if isinstance(df, pl.DataFrame) else df
else:
message = "Unable to find and delete PL/Budget rows"
paramDict = add_warning_message_in_period_options_tab(paramDict, message)
message = "PL/Budget rows must be named with one of the following stems in upper or lower case"
paramDict = add_warning_message_in_period_options_tab(paramDict, message)
paramDict = add_write_message_in_period_options_tab(
paramDict, planStemArray
)
message = (
"Rename PL/Budget rows or set Compare widget to Scenarios and rerun"
)
paramDict = add_warning_message_in_period_options_tab(paramDict, message)
return dfCopy.lazy() if isinstance(dfCopy, pl.DataFrame) else dfCopy
else:
return dfCopy.lazy() if isinstance(dfCopy, pl.DataFrame) else dfCopy
def set_dataframe_to_compare_periods(dfCopy, paramDict, chartDict):
namingParams = get_naming_params()
comparePeriods = namingParams["comparePeriods"]
compareScenariosOrPeriods = namingParams["compareScenariosOrPeriods"]
periodName = namingParams["periodName"]
if (
compareScenariosOrPeriods in chartDict
and chartDict[compareScenariosOrPeriods] == comparePeriods
):
df = clean_PL_rows(dfCopy, paramDict, chartDict)
return df
else:
return dfCopy
def check_if_date_column_exists(df, paramDict, chartDict):
"""Return a dataset for time filtering.
When a date column is present we duplicate ``df`` so date filters can be
applied independently of the main pipeline. If no date column exists but a
period column does, a duplicate is still returned to allow period based
operations to continue. Otherwise an empty ``LazyFrame`` is produced.
"""
namingParams = get_naming_params()
dateColFound = namingParams["dateColFound"]
periodColFound = namingParams["periodColFound"]
filterDates = namingParams["filterDates"]
metConditionValue = namingParams["metConditionValue"]
if dateColFound in paramDict and paramDict[dateColFound]:
if (
not paramDict[periodColFound]
or filterDates in chartDict
and chartDict[filterDates] == metConditionValue
):
dfDates = duplicate_dataframe(df)
else:
dfDates = pl.LazyFrame()
elif paramDict.get(periodColFound):
dfDates = duplicate_dataframe(df)
else:
dfDates = pl.LazyFrame()
return dfDates
def filter_dataframe_in_date_range(df, chartDict):
"""
filtering the dataframe based on the chosen date range
"""
namingParams = get_naming_params()
dateName = namingParams["dateName"]
dateRangeArray = namingParams["dateRangeArray"]
filterDates = namingParams["filterDates"]
if filterDates in chartDict and chartDict[filterDates]:
dateRangeArray = chartDict[dateRangeArray]
min_date = pl.Series([dateRangeArray[0]]).str.to_datetime()[0]
max_date = pl.Series([dateRangeArray[1]]).str.to_datetime()[0]
if min_date < max_date:
if not isinstance(df, (pl.DataFrame, pl.LazyFrame)):
df = pl.DataFrame(df)
df = df.filter(
(pl.col(dateName) >= min_date) & (pl.col(dateName) <= max_date)
)
return df
def get_comparison_period(paramDict, chartDict):
"""
we somehow get the time metric (by month, by year, by week,by quarter, by day)
on how to split the dataset along time
"""
namingParams = get_naming_params()
datePeriodName = namingParams["datePeriodName"]
weekName = namingParams["weekName"]
monthName = namingParams["monthName"]
quarterName = namingParams["quarterName"]
yearName = namingParams["yearName"]
periodChoice = namingParams["periodChoice"]
metricDict = {
"W": weekName,
"M": monthName,
"Q": quarterName,
"Y": yearName,
}
if periodChoice in chartDict and chartDict[periodChoice]:
paramDict[datePeriodName] = chartDict[periodChoice]
else:
paramDict[datePeriodName] = None
return paramDict
def calculate_fiscal_year_lazy(date_col, start_month):
"""Calculate fiscal year based on the start month."""
return (
pl.when(pl.col(date_col).dt.month() >= start_month)
.then(pl.col(date_col).dt.year() + 1)
.otherwise(pl.col(date_col).dt.year())
)
def calculate_fiscal_quarter_lazy(month_col, start_month):
"""Calculate fiscal quarter based on the start month."""
shifted_month = (month_col - start_month) % 12 + 1
return ((shifted_month - 1) // 3 + 1).alias("FQ")
def ensure_datetime_column(lf, date_name, date_format):
"""
Ensure the specified column is in Datetime format.
If not, convert it from string using the given date format.
"""
columns, schema = utils.get_schema_and_column_names(lf)
# Check the schema to determine the column's data type
if schema[date_name] != pl.Datetime:
# If not Datetime, cast to string and parse as Datetime
lf = lf.with_columns(
pl.col(date_name)
.cast(pl.Utf8)
.str.strptime(pl.Datetime, format=date_format, strict=False)
.alias(date_name)
)
return lf
def detect_date_format(
lf: pl.LazyFrame, date_name: str, sample_size: int = 5, default: str = "%Y-%m-%d"
) -> str:
"""Infer the date format for ``date_name`` by sampling a few rows."""
_, schema = utils.get_schema_and_column_names(lf)
dtype = schema.get(date_name) if schema else None
if dtype in (pl.Datetime, pl.Date):
return default
try:
sample = (
lf.select(pl.col(date_name).drop_nulls())
.limit(sample_size)
.collect(engine="streaming")
)
except Exception as e:
try: # pragma: no cover - logging only
notifier.error("Something went wrong.")
except Exception as e: # pragma: no cover - notifier unavailable
logging.exception(e)
return default
candidates = ["%Y-%m-%d", "%m/%d/%Y", "%m-%d-%Y", "%d/%m/%Y", "%d-%m-%Y"]
for fmt in candidates:
parsed = sample.select(
pl.col(date_name).str.strptime(pl.Date, fmt, strict=False)
)
if parsed.filter(pl.col(date_name).is_null()).height == 0:
return fmt
return default
def assign_fiscal_period(df, paramDict, chartDict):
namingParams = get_naming_params()
dateName = namingParams["dateName"]
dateFormat = detect_date_format(df, dateName)
periodName = namingParams["periodName"]
fiscalStartMonth = namingParams["fiscalStartMonth"]
quarterName = namingParams["quarterName"]
yearName = namingParams["yearName"]
datePeriodName = namingParams["datePeriodName"]
startMonth = chartDict[fiscalStartMonth]
period = paramDict[datePeriodName]
# Ensure the date column is in datetime format
# Ensure the column is a string before applying .str.strptime
df = ensure_datetime_column(df, dateName, dateFormat)
if period == yearName:
# Calculate fiscal year
df = df.with_columns(
calculate_fiscal_year_lazy(dateName, startMonth).alias(periodName)
).with_columns(
pl.format("’FY{}", pl.col(periodName).cast(pl.Utf8).str.slice(-2)).alias(
periodName
)
)
elif period == quarterName:
# Calculate fiscal quarter
df = df.with_columns(
calculate_fiscal_year_lazy(dateName, startMonth).alias("FY"),
calculate_fiscal_quarter_lazy(pl.col(dateName).dt.month(), startMonth),
).with_columns(
pl.format(
"’{}’FQ{}", pl.col("FY").cast(pl.Utf8).str.slice(-2), pl.col("FQ")
).alias(periodName)
)
df = drop_columns(df, ["FY", "FQ"])
return df
def convert_date_to_period(df, paramDict, chartDict):
"""
if we find a date column, we convert it into period based on the requested period
"""
namingParams = get_naming_params()
weekName = namingParams["weekName"]
monthName = namingParams["monthName"]
quarterName = namingParams["quarterName"]
yearName = namingParams["yearName"]
periodName = namingParams["periodName"]
datePeriodName = namingParams["datePeriodName"]
dateName = namingParams["dateName"]
periodColFound = namingParams["periodColFound"]
metConditionValue = namingParams["metConditionValue"]
notMetConditionValue = namingParams["notMetConditionValue"]
fiscalYear = namingParams["fiscalYear"]
period = paramDict[datePeriodName]
# Ensure date column is in Datetime format before applying .dt operations
date_format = detect_date_format(df, dateName)
df = ensure_datetime_column(df, dateName, date_format)
if 1 == 1: # try:
if (
fiscalYear in chartDict
and chartDict[fiscalYear]
and period in [quarterName, yearName]
):
df = assign_fiscal_period(df, paramDict, chartDict)
else:
if period == weekName:
df = df.with_columns(
# Combine formatting and prefix in one operation
pl.format(
"’{}",
pl.col(dateName).dt.strftime("%y-W%V"), # Format as year-week
).alias(periodName)
)
elif period == monthName:
df = df.with_columns(
[
# Compute month and year inline
pl.format(
"’{}-{}",
pl.col(dateName)
.dt.year()
.cast(pl.Utf8)
.str.slice(-2), # Year (last 2 digits)
pl.col(dateName)
.dt.month()
.cast(pl.Utf8)
.str.zfill(2), # Month (2 digits)
).alias(periodName)
]
)
elif period == quarterName:
df = df.with_columns(
pl.format(
"’{}-Q{}",
pl.col(dateName).dt.strftime(
"%y"
), # Last two digits of the year
((pl.col(dateName).dt.month() - 1) // 3 + 1).cast(
pl.Utf8
), # Calculate quarter
).alias(periodName)
)
elif period == yearName:
df = df.with_columns(
pl.format(
"’{}", pl.col(dateName).dt.year().cast(pl.Utf8).str.slice(-2)
).alias(periodName)
)
else:
df = df.with_columns(
pl.format(
"’{}", pl.col(dateName).dt.year().cast(pl.Utf8).str.slice(-2)
).alias(periodName)
)
paramDict[datePeriodName] = yearName
paramDict[periodColFound] = metConditionValue
else: # except:
paramDict[periodColFound] = notMetConditionValue
return df, paramDict
def change_date_field_to_relevant_period(df, paramDict, chartDict):
"""
we convert the date column (if it exists) in a period column.
we look for a period column in the date column does not exist
If we found a period column, we delete the data columns
"""
namingParams = get_naming_params()
periodName = namingParams["periodName"]
dateName = namingParams["dateName"]
likelyPeriodCols = namingParams["likelyPeriodCols"]
periodColFound = namingParams["periodColFound"]
numericTypesArray = [
pl.Int8,
pl.Int16,
pl.Int32,
pl.Int64,
pl.UInt8,
pl.UInt16,
pl.UInt32,
pl.UInt64,
pl.Float32,
pl.Float64,
]
columns, schema = utils.get_schema_and_column_names(df)
if dateName in columns and not paramDict[periodColFound]:
df, paramDict = convert_date_to_period(df, paramDict, chartDict)
columns, schema = utils.get_schema_and_column_names(df)
if paramDict[periodColFound]:
df = df.filter(pl.col(periodName).is_not_null())
columns, schema = utils.get_schema_and_column_names(df)
if periodName in columns and schema.get(periodName) in numericTypesArray:
df = df.with_columns(pl.col(periodName).cast(pl.Int32).alias(periodName))
return df, paramDict
def check_if_can_process_dates(df, paramDict):
namingParams = get_naming_params()
impossibleToProcessFile = namingParams["impossibleToProcessFile"]
dateColFound = namingParams["dateColFound"]
notMetConditionValue = namingParams["notMetConditionValue"]
metConditionValue = namingParams["metConditionValue"]
impossibleToProcessFile = paramDict[impossibleToProcessFile]
isDateColFound = paramDict[dateColFound]
processDates = notMetConditionValue
if not impossibleToProcessFile and utils.is_valid_lazyframe(df) and isDateColFound:
processDates = metConditionValue
return processDates
def set_date_parameters(df, paramDict):
namingParams = get_naming_params()
notMetConditionValue = namingParams["notMetConditionValue"]
compareWithYearBefore = notMetConditionValue
paramDict, mostRecentDate, leastRecentDate, periodLengthInMonths = (
get_period_length(df, paramDict, False)
)
periodLengthInMonths = int(round(periodLengthInMonths, 0))
mostRecentMonth = mostRecentDate.strftime("%B")
return compareWithYearBefore, paramDict, periodLengthInMonths, mostRecentMonth
def manage_date_filters(dfDates, chartDict, paramDict, periodLengthInMonths):
namingParams = get_naming_params()
filterDates = namingParams["filterDates"]
if filterDates in chartDict and chartDict[filterDates]:
(
copyDict,
mostRecentDatesDate,
leastRecentDatesDate,
periodDatesLengthInMonths,
) = get_period_length(dfDates, paramDict, True)
periodDatesLengthInMonths = int(round(periodLengthInMonths, 0))
numberOfMonths = max(periodDatesLengthInMonths, periodLengthInMonths)
else:
numberOfMonths = periodLengthInMonths
return numberOfMonths
def check_if_more_than_twenty_four_months(paramDict, numberOfMonths):
namingParams = get_naming_params()
notMetConditionValue = namingParams["notMetConditionValue"]
metConditionValue = namingParams["metConditionValue"]
twentyfourMonthsInDataset = namingParams["twentyfourMonthsInDataset"]
paramDict[twentyfourMonthsInDataset] = notMetConditionValue
if numberOfMonths >= 24:
paramDict[twentyfourMonthsInDataset] = metConditionValue
return paramDict
def check_compare_with_year_before(chartDict, paramDict, periodLengthInMonths):
namingParams = get_naming_params()
compareWithYearBeforeKey = namingParams["compareWithYearBefore"]
setTimePeriodTabKey = namingParams["setTimePeriodTab"]
datePeriodName = namingParams["datePeriodName"]
monthName = namingParams["monthName"]
quarterName = namingParams["quarterName"]
notMetConditionValue = namingParams["notMetConditionValue"]
infoMessageType = namingParams["infoMessageType"]
setTimePeriodTabLabel = namingParams["setTimePeriodTabLabel"]
if compareWithYearBeforeKey in chartDict and chartDict[compareWithYearBeforeKey]:
if (
datePeriodName in paramDict
and paramDict[datePeriodName] == monthName
and periodLengthInMonths <= 13
) or (
datePeriodName in paramDict
and paramDict[datePeriodName] == quarterName
and periodLengthInMonths <= 23
):
chartDict[compareWithYearBeforeKey] = notMetConditionValue
message = (
"Only "
+ str(periodLengthInMonths)
+ " months in dataset. 'Compare with rolling period' set to False"
)
paramDict = add_app_message_to_paramdict(
message,
infoMessageType,
setTimePeriodTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=1,
)
return chartDict, paramDict
def process_yearly_aggregation(df, dfDates, paramDict, chartDict):
"""
we want to compare the most recent 12 months to the 12 months of the year before
"""
processDates = check_if_can_process_dates(df, paramDict)
periodLengthInMonths, mostRecentMonth = False, False
if processDates:
compareWithYearBefore, paramDict, periodLengthInMonths, mostRecentMonth = (
set_date_parameters(df, paramDict)
)
numberOfMonths = manage_date_filters(
dfDates, chartDict, paramDict, periodLengthInMonths
)
paramDict = check_if_more_than_twenty_four_months(paramDict, numberOfMonths)
chartDict, paramDict = check_compare_with_year_before(
chartDict, paramDict, periodLengthInMonths
)
return chartDict, paramDict, periodLengthInMonths, mostRecentMonth, processDates
def find_columns_and_manage_dates(df, paramDict, chartDict):
"""
runs profiler and finds value and date columns
"""
namingParams = get_naming_params()
errorMessageType = namingParams["errorMessageType"]
loadDataTabKey = namingParams["loadDataTab"]
df, paramDict = check_for_string_metric_cols(df, paramDict)
df, paramDict = find_volume_columns(df, paramDict)
df, paramDict = find_monetary_columns(df, paramDict)
df, paramDict = find_discount_and_cogs_columns(df, paramDict)
df, paramDict = find_distribution_metric_columns(df, paramDict)
df, paramDict = find_checkout_columns(df, paramDict)
df, paramDict = find_visits_columns(df, paramDict)
df = delete_rows_with_all_zeroes(df, paramDict)
dfPlan = pl.LazyFrame()
if utils.is_valid_lazyframe(df):
df = set_dataframe_to_compare_periods(df, paramDict, chartDict)
dfDates = check_if_date_column_exists(df, paramDict, chartDict)
df = filter_dataframe_in_date_range(df, chartDict)
dfDates = filter_dataframe_in_date_range(dfDates, chartDict)
paramDict = get_comparison_period(paramDict, chartDict)
df, paramDict = change_date_field_to_relevant_period(df, paramDict, chartDict)
chartDict, paramDict, periodLengthInMonths, mostRecentMonth, processDates = (
process_yearly_aggregation(df, dfDates, paramDict, chartDict)
)
else:
dfDates = pl.LazyFrame()
dfPlan = pl.LazyFrame()
message = "Empty dataset, unable to process"
paramDict = add_app_message_to_paramdict(
message,
errorMessageType,
loadDataTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=0,
)
periodLengthInMonths, mostRecentMonth, processDates = None, None, None
return (
df,
dfDates,
paramDict,
chartDict,
periodLengthInMonths,
mostRecentMonth,
processDates,
)
def change_time_aggregation(df, paramDict, chartDict, newTimeAggregation, automateDict):
"""
if we have only one year in the dataframe the time aggregation parameter cannot be year. So we force it to month
"""
namingParams = get_naming_params()
periodName = namingParams["periodName"]
datePeriodName = namingParams["datePeriodName"]
monthName = namingParams["monthName"]
yearName = namingParams["yearName"]
quarterName = namingParams["quarterName"]
dateName = namingParams["dateName"]
workColumn = namingParams["workColumn"]
periodColFound = namingParams["periodColFound"]
filterDates = namingParams["filterDates"]
metConditionValue = namingParams["metConditionValue"]
compareWithYearBeforeKey = namingParams["compareWithYearBefore"]
periodToDateKey = namingParams["periodToDate"]
notMetConditionValue = namingParams["notMetConditionValue"]
metConditionValue = namingParams["metConditionValue"]
changeAggregationPeriod = namingParams["changeAggregationPeriod"]
changedTimeAggregation = namingParams["changedTimeAggregation"]
from modules.layout.set_up_widgets import get_period_aggregation_change_choice
message = notMetConditionValue
periods = get_periods_array(df)
paramDict, mostRecentDate, leastRecentDate, periodLengthInMonths = (
get_period_length(df, paramDict, False)
)
if paramDict[datePeriodName] == yearName:
if not (
filterDates in chartDict and chartDict[filterDates] == metConditionValue
):
if newTimeAggregation == monthName:
chartDict = get_period_aggregation_change_choice(
chartDict, monthName, automateDict
)
if chartDict[changeAggregationPeriod]:
message = (
"Dataset contains "
+ str(int(round(periodLengthInMonths, 0)))
+ " months which is less than 13 months."
)
paramDict[datePeriodName] = monthName
chartDict[compareWithYearBeforeKey] = notMetConditionValue
chartDict[periodToDateKey] = notMetConditionValue
df = df.with_columns(
[
pl.col(dateName)
.dt.month()
.cast(pl.Utf8)
.str.zfill(2)
.alias(periodName),
pl.col(dateName).dt.year().alias(workColumn),
]
).with_columns(
pl.concat_str(
[
pl.lit("’"),
pl.col(workColumn).cast(pl.Utf8).str.slice(-2),
pl.lit("-"),
pl.col(periodName),
]
).alias(periodName)
)
message = (
message
+ " 'Date Aggregation' parameter set to month. You can set the parameter back to year in the Period Options tab."
)
paramDict[changedTimeAggregation] = metConditionValue
elif newTimeAggregation == quarterName:
chartDict = get_period_aggregation_change_choice(
chartDict, monthName, automateDict
)
if chartDict[changeAggregationPeriod]:
message = (
"Dataset contains "
+ str(int(round(periodLengthInMonths, 0)))
+ " months which is less than 24 months."
)
paramDict[datePeriodName] = quarterName
chartDict[compareWithYearBeforeKey] = notMetConditionValue
chartDict[periodToDateKey] = notMetConditionValue
df = df.with_columns(
pl.format(
"’{}-Q{}",
pl.col(dateName).dt.strftime("%y"),
((pl.col(dateName).dt.month() - 1) // 3 + 1).cast(pl.Utf8),
).alias(periodName)
)
message = (
message
+ " 'Date Aggregation' parameter set to quarter. You can set the parameter back to year in the Period Options tab."
)
paramDict[changedTimeAggregation] = metConditionValue
if message:
paramDict = add_warning_message_in_period_options_tab(
paramDict, message
)
df = drop_columns(df, [workColumn])
return df, paramDict, chartDict
def show_message_most_recent_month_not_december(mostRecentMonth, paramDict, chartDict):
namingParams = get_naming_params()
mostRecentPeriod = namingParams["mostRecentPeriod"]
setTimePeriodTabKey = namingParams["setTimePeriodTab"]
warningMessageType = namingParams["warningMessageType"]
setTimePeriodTabLabel = namingParams["setTimePeriodTabLabel"]
fiscalYear = namingParams["fiscalYear"]
if fiscalYear in chartDict and chartDict[fiscalYear]:
pass
elif mostRecentPeriod in chartDict and chartDict[mostRecentPeriod] == -1:
message = (
"Comparison set on 'Calendar year' but most recent month "
+ mostRecentMonth
+ ", not December."
)
paramDict = add_app_message_to_paramdict(
message,
warningMessageType,
setTimePeriodTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=1,
)
message = "You might be comparing periods of different length."
paramDict = add_app_message_to_paramdict(
message,
warningMessageType,
setTimePeriodTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=1,
)
message = "Consider changing setting to 'Year-to-Date' or to 'Rolling Periods'."
paramDict = add_app_message_to_paramdict(
message,
warningMessageType,
setTimePeriodTabKey,
paramDict,
isMessage=True,
isToast=False,
colNumber=1,
)
message = (
"Consider changing setting to 'Year-to-Date' or to 'Rolling Periods' in the "
+ setTimePeriodTabLabel
+ " tab."
)
paramDict = add_app_message_to_paramdict(
message,
warningMessageType,
setTimePeriodTabKey,
paramDict,
isMessage=False,
isToast=True,
colNumber=1,
)
return paramDict
def show_message_less_than_twenty_four_months(periodLengthInMonths, paramDict):
namingParams = get_naming_params()
datePeriodName = namingParams["datePeriodName"]
quarterName = namingParams["quarterName"]
monthName = namingParams["monthName"]
yearName = namingParams["yearName"]
infoMessageType = namingParams["infoMessageType"]
setTimePeriodTabKey = namingParams["setTimePeriodTab"]
if (
datePeriodName in paramDict
and paramDict[datePeriodName] == yearName
and periodLengthInMonths < 24
):
message = (
"Dataset contains only "
+ str(periodLengthInMonths)
+ " months. Rolling year calculation requires 24 months."
)
paramDict = add_app_message_to_paramdict(
message,
infoMessageType,
setTimePeriodTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=0,
)
message = "You might be comparing periods of different length. Consider changing setting to 'Year-to-Date'."
paramDict = add_app_message_to_paramdict(
message,
infoMessageType,
setTimePeriodTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=0,
)
if (
datePeriodName in paramDict
and paramDict[datePeriodName] == quarterName
and periodLengthInMonths < 15
):
message = (
"Dataset contains only "
+ str(periodLengthInMonths)
+ " months. Rolling quarter calculation requires 15 months."
)
paramDict = add_app_message_to_paramdict(
message,
infoMessageType,
setTimePeriodTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=0,
)
message = "You might be comparing periods of different length. Consider changing setting to 'Quarter-to-Date'."
paramDict = add_app_message_to_paramdict(
message,
infoMessageType,
setTimePeriodTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=0,
)
if (
datePeriodName in paramDict
and paramDict[datePeriodName] == monthName
and periodLengthInMonths < 13
):
message = (
"Dataset contains only "
+ str(periodLengthInMonths)
+ " months. Rolling month calculation requires 13 months."
)
paramDict = add_app_message_to_paramdict(
message,
infoMessageType,
setTimePeriodTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=0,
)
message = "You might be comparing periods of different length. Consider changing setting to 'Month-to-Date'."
paramDict = add_app_message_to_paramdict(
message,
infoMessageType,
setTimePeriodTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=0,
)
return paramDict
def check_if_date_aggregation_is_year(
df,
chartDict,
paramDict,
periodLengthInMonths,
mostRecentMonth,
automateDict,
processDates,
):
namingParams = get_naming_params()
datePeriodName = namingParams["datePeriodName"]
yearName = namingParams["yearName"]
monthName = namingParams["monthName"]
quarterName = namingParams["quarterName"]
decemberName = namingParams["decemberName"]
periodToDateKey = namingParams["periodToDate"]
compareWithYearBeforeKey = namingParams["compareWithYearBefore"]
canPlotYearToYear = namingParams["canPlotYearToYear"]
notMetConditionValue = namingParams["notMetConditionValue"]
if processDates:
if datePeriodName in paramDict and paramDict[datePeriodName] == yearName:
if periodLengthInMonths <= 13:
df, paramDict, chartDict = change_time_aggregation(
df, paramDict, chartDict, monthName, automateDict
)
elif periodLengthInMonths <= 23 and not chartDict[periodToDateKey]:
df, paramDict, chartDict = change_time_aggregation(
df, paramDict, chartDict, quarterName, automateDict
)
elif mostRecentMonth != decemberName:
if (
not chartDict[compareWithYearBeforeKey]
and not chartDict[periodToDateKey]
):
paramDict = show_message_most_recent_month_not_december(
mostRecentMonth, paramDict, chartDict
)
if not chartDict[compareWithYearBeforeKey]:
chartDict[canPlotYearToYear] = notMetConditionValue
if (
compareWithYearBeforeKey in chartDict
and chartDict[compareWithYearBeforeKey]
):
paramDict = show_message_less_than_twenty_four_months(
periodLengthInMonths, paramDict
)
return chartDict, paramDict
def calculate_periods(
df,
paramDict,
chartDict,
tyYaDatesArray,
session_manager: SessionManager | None = None,
):
namingParams = get_naming_params()
session_manager = session_manager or SessionManager()
datePeriodName = namingParams["datePeriodName"]
yearName = namingParams["yearName"]
compareWithYearBeforeKey = namingParams["compareWithYearBefore"]
periodToDateKey = namingParams["periodToDate"]
if (
paramDict[datePeriodName] == yearName
and compareWithYearBeforeKey in chartDict
and chartDict[compareWithYearBeforeKey]
):
df, paramDict, tyYaDatesArray, chartDict = calculate_rolling_period(
df, tyYaDatesArray, paramDict, chartDict
)
if periodToDateKey in chartDict and chartDict[periodToDateKey]:
df, paramDict = calculate_period_to_date_year_ago(
df, paramDict, session_manager=session_manager
)
elif periodToDateKey in chartDict and chartDict[periodToDateKey]:
if paramDict[datePeriodName] == yearName:
df, paramDict = calculate_period_to_date_year_ago(
df, paramDict, session_manager=session_manager
)
elif (
compareWithYearBeforeKey in chartDict
and chartDict[compareWithYearBeforeKey]
):
df, paramDict = calculate_period_to_date_year_ago(
df, paramDict, session_manager=session_manager
)
else:
df, paramDict = calculate_period_to_date_same_year(
df, paramDict, session_manager=session_manager
)
return df, paramDict, tyYaDatesArray, chartDict
def group_this_year_and_year_ago(
dfCopy,
paramDict,
chartDict,
processDates,
session_manager: SessionManager | None = None,
preserve_date_col: bool = False,
):
"""Compare the latest 12 months to the same period a year ago.
Parameters
----------
dfCopy: pl.LazyFrame | pl.DataFrame
Source dataframe.
paramDict: dict
Parameter dictionary used across data preparation.
chartDict: dict
Chart settings used by downstream modules.
processDates: bool
Whether the dataframe already contains calculated date periods.
session_manager: SessionManager | None, optional
Session manager instance.
preserve_date_col: bool, optional
When ``True`` the original date column is kept in ``df``.
"""
namingParams = get_naming_params()
tyYaDates = namingParams["tyYaDates"]
dateName = namingParams["dateName"]
tyYaDatesArray = []
session_manager = session_manager or SessionManager()
df = duplicate_dataframe(dfCopy)
if processDates:
df, paramDict, tyYaDatesArray, chartDict = calculate_periods(
df, paramDict, chartDict, tyYaDatesArray, session_manager=session_manager
)
sortedtyYaDatesArray = sorted(tyYaDatesArray)
paramDict[tyYaDates] = sortedtyYaDatesArray
dfPlan = duplicate_dataframe(df)
if not preserve_date_col:
df = drop_columns(df, [dateName])
return df, dfPlan, paramDict
def apply_filter_and_copy_df(df, selectedPeriodsArray, paramDict, chartDict):
"""
if the users has requested comparison to the corresponding period of the year before, we check
if we have that period in the datased
"""
namingParams = get_naming_params()
periodName = namingParams["periodName"]
fcName = namingParams["fcName"]
compareScenariosOrPeriods = namingParams["compareScenariosOrPeriods"]
compareScenarios = namingParams["compareScenarios"]
periodsArray = copy.deepcopy(selectedPeriodsArray)
if (
compareScenariosOrPeriods in chartDict
and chartDict[compareScenariosOrPeriods] == compareScenarios
):
# Duplicate the dataframe (lazy equivalent)
dfPeriods = df.clone()
# Filter df by periodsArray
df = df.filter(pl.col(periodName).is_in(periodsArray))
# Exclude rows where periodName is fcName
df = df.filter(~pl.col(periodName).is_in([fcName]))
# Append fcName to periodsArray if not present
if fcName not in periodsArray:
periodsArray.append(fcName)
# Filter dfPeriods by the updated periodsArray
dfPeriods = dfPeriods.filter(pl.col(periodName).is_in(periodsArray))
else:
# Filter df by periodsArray
df = df.filter(pl.col(periodName).is_in(periodsArray))
# Duplicate after filtering
dfPeriods = df.clone()
# Exclude rows where periodName is fcName
df = df.filter(~pl.col(periodName).is_in([fcName]))
return df, dfPeriods
def check_if_year_before_period_exists(sortedPeriods, mostRecentPeriodIndex, paramDict):
"""
if the users has requested comparison to the corresponding period of the year before, we check
if we have that period in the datased
"""
namingParams = get_naming_params()
datePeriodName = paramDict[namingParams["datePeriodName"]]
weekName = namingParams["weekName"]
monthName = namingParams["monthName"]
quarterName = namingParams["quarterName"]
metricDict = {
weekName: 51,
monthName: 11,
quarterName: 3,
}
periods = copy.deepcopy(sortedPeriods)
periods = periods[0:mostRecentPeriodIndex]
if len(periods) >= metricDict[datePeriodName]:
yearBeforePeriodIndex = metricDict[datePeriodName]
else:
yearBeforePeriodIndex = 0
return yearBeforePeriodIndex
def transform_integer_periods(df, periodsArray, paramDict):
"""
when importing from excel the period column is parsed as a huge integer
"""
namingParams = get_naming_params()
uploadedFileType = namingParams["uploadedFileType"]
periodName = namingParams["periodName"]
columns, schema = utils.get_schema_and_column_names(df)
if uploadedFileType in paramDict and paramDict[uploadedFileType] == "xlsx":
if schema.get(periodName) == pl.Datetime:
try:
cleanedperiodsArray = []
for element in periodsArray:
element = dt.datetime.fromtimestamp(int(element) / 1e9)
cleanedperiodsArray.append(str(element.date()))
periodsArray = cleanedperiodsArray
df = df.with_columns(
pl.col(periodName).dt.strftime("%Y-%m-%d").alias(periodName)
)
except Exception as e:
logging.exception(e)
exc = print_error_details(e)
errorMessageType = namingParams["errorMessageType"]
loadDataTabKey = namingParams["loadDataTab"]
paramDict = add_app_message_to_paramdict(
e,
errorMessageType,
loadDataTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=0,
)
df = df.with_columns(pl.col(periodName).cast(pl.Utf8).alias(periodName))
else:
df = df.with_columns(pl.col(periodName).cast(pl.Utf8).alias(periodName))
return df, periodsArray
def check_if_year_before_previous_year(paramDict, chartDict):
namingParams = get_naming_params()
dateColFound = namingParams["dateColFound"]
datePeriodName = namingParams["datePeriodName"]
selectedPeriods = namingParams["selectedPeriods"]
notMetConditionValue = namingParams["notMetConditionValue"]
metConditionValue = namingParams["metConditionValue"]
isYearBeforePy = namingParams["isYearBeforePy"]
paramDict[isYearBeforePy] = notMetConditionValue
if (dateColFound not in paramDict and datePeriodName not in paramDict) or not (
paramDict[dateColFound] and not paramDict[datePeriodName]
):
if selectedPeriods in chartDict:
if (
len(chartDict[selectedPeriods][0]) == 5
and len(chartDict[selectedPeriods][1]) == 5
):
firstPeriod = chartDict[selectedPeriods][0][1:]
secondPeriod = chartDict[selectedPeriods][1][1:]
try:
firstPeriod = int(firstPeriod)
secondPeriod = int(secondPeriod)
except Exception as e:
logging.exception(e)
exc = print_error_details(e)
errorMessageType = namingParams["errorMessageType"]
loadDataTabKey = namingParams["loadDataTab"]
paramDict = add_app_message_to_paramdict(
e,
errorMessageType,
loadDataTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=0,
)
firstPeriod, secondPeriod = 0, 0
if secondPeriod - firstPeriod >= 2:
paramDict[isYearBeforePy] = metConditionValue
elif (
len(chartDict[selectedPeriods][0]) == 4
and len(chartDict[selectedPeriods][1]) == 4
):
try:
firstPeriod = int(chartDict[selectedPeriods][0])
secondPeriod = int(chartDict[selectedPeriods][1])
except Exception as e:
logging.exception(e)
exc = print_error_details(e)
errorMessageType = namingParams["errorMessageType"]
loadDataTabKey = namingParams["loadDataTab"]
paramDict = add_app_message_to_paramdict(
e,
errorMessageType,
loadDataTabKey,
paramDict,
isMessage=True,
isToast=True,
colNumber=0,
)
firstPeriod, secondPeriod = 0, 0
if secondPeriod - firstPeriod >= 2:
paramDict[isYearBeforePy] = metConditionValue
return paramDict
def filter_out_useless_periods(df, paramDict, chartDict):
"""
if we have a period column and it has more than two values we want to keep only
two periods. Conventionally, we keep the two most recent periods
"""
namingParams = get_naming_params()
periodName = namingParams["periodName"]
periodColFound = namingParams["periodColFound"]
allPeriodsList = namingParams["allPeriodsList"]
selectedPeriods = namingParams["selectedPeriods"]
numberOfPeriodsFound = namingParams["numberOfPeriodsFound"]
mostRecentPeriod = namingParams["mostRecentPeriod"]
compareWithYearBefore = namingParams["compareWithYearBefore"]
datePeriodName = namingParams["datePeriodName"]
periodName = namingParams["periodName"]
yearName = namingParams["yearName"]
fcName = namingParams["fcName"]
plName = namingParams["plName"]
currentYearName = namingParams["currentYearName"]
notMetConditionValue = namingParams["notMetConditionValue"]
metConditionValue = namingParams["metConditionValue"]
tyYaDates = namingParams["tyYaDates"]
periodToDate = namingParams["periodToDate"]
changedTimeAggregation = namingParams["changedTimeAggregation"]
isPeriodColFound = paramDict[periodColFound]
applyFilter = False
selectedPeriodsArray = []
columns, schema = utils.get_schema_and_column_names(df)
if isPeriodColFound or periodName in columns:
paramDict[periodColFound] = isPeriodColFound
if periodName in columns:
df = df.with_columns(pl.col(periodName).cast(pl.Utf8).alias(periodName))
reverseSort = False
periods = get_periods_array(df)
if fcName in periods:
periods.remove(fcName)
if plName in periods:
reverseSort = True
if compareWithYearBefore in chartDict and chartDict[compareWithYearBefore]:
if (
datePeriodName in paramDict
and paramDict[datePeriodName] == yearName
and currentYearName in periods
):
reverseSort = True
chosenDatePeriodName = paramDict[datePeriodName]
periods = [str(i) for i in periods]
sortedPeriods = sorted(periods, reverse=reverseSort)
numberOfPeriods = len(periods)
paramDict[numberOfPeriodsFound] = numberOfPeriods
if numberOfPeriods > 1 and mostRecentPeriod in chartDict:
mostRecentPeriod = chartDict[mostRecentPeriod]
mostRecentPeriodIndex, paramDict = get_dataset_specific_parameter(
paramDict, namingParams["mostRecentPeriodIndex"], False
)
if numberOfPeriods > abs(mostRecentPeriod):
mostRecentPeriodIndex = mostRecentPeriod
else:
mostRecentPeriodIndex = -(numberOfPeriods - 1)
yearBeforePeriodIndex = 0
if compareWithYearBefore in chartDict and chartDict[compareWithYearBefore]:
if datePeriodName in paramDict and paramDict[datePeriodName]:
if paramDict[datePeriodName] != yearName:
yearBeforePeriodIndex = check_if_year_before_period_exists(
sortedPeriods, mostRecentPeriodIndex, paramDict
)
otherPeriodIndex = mostRecentPeriodIndex - yearBeforePeriodIndex - 1
if len(sortedPeriods) < abs(otherPeriodIndex):
otherPeriodIndex = mostRecentPeriodIndex - 1
if periodToDate not in chartDict or not chartDict[periodToDate]:
chartDict[compareWithYearBefore] = notMetConditionValue
paramDict[changedTimeAggregation] = metConditionValue
message = "Dataset contains less than 24 months of data. 'Date Aggregation' parameter set to month, 'Compare to Rolling Period' parameter forced to False. You can set the parameter back to year in the Period Options tab. "
add_info_message_in_period_options_tab(paramDict, message)
df, sortedPeriods = transform_integer_periods(df, sortedPeriods, paramDict)
selectedPeriodsArray = [
sortedPeriods[otherPeriodIndex],
sortedPeriods[mostRecentPeriodIndex],
]
chartDict[mostRecentPeriod] = mostRecentPeriodIndex
paramDict[allPeriodsList] = sortedPeriods
paramDict[selectedPeriods] = selectedPeriodsArray
chartDict[selectedPeriods] = selectedPeriodsArray
dfAllPeriods = duplicate_dataframe(df)
dfAllPeriods = dfAllPeriods.sort(by=periodName, descending=False)
df, dfPeriods = apply_filter_and_copy_df(
df, selectedPeriodsArray, paramDict, chartDict
)
paramDict = check_if_year_before_previous_year(paramDict, chartDict)
return df, dfPeriods, dfAllPeriods, paramDict, chartDict
elif numberOfPeriods == 1 and mostRecentPeriod in chartDict:
df, sortedPeriods = transform_integer_periods(df, sortedPeriods, paramDict)
selectedPeriodsArray = sortedPeriods
mostRecentPeriodIndex = 0
chartDict[mostRecentPeriod] = mostRecentPeriodIndex
paramDict[allPeriodsList] = sortedPeriods
paramDict[selectedPeriods] = selectedPeriodsArray
chartDict[selectedPeriods] = selectedPeriodsArray
dfAllPeriods = duplicate_dataframe(df)
df, dfPeriods = apply_filter_and_copy_df(
df, selectedPeriodsArray, paramDict, chartDict
)
return df, dfPeriods, dfAllPeriods, paramDict, chartDict
else:
dfPeriods, dfAllPeriods = pl.LazyFrame(), pl.LazyFrame()
return df, dfPeriods, dfAllPeriods, paramDict, chartDict
else:
dfPeriods, dfAllPeriods = pl.LazyFrame(), pl.LazyFrame()
return df, dfPeriods, dfAllPeriods, paramDict, chartDict
SHA-256: 99503c3b78f9292a3886c0dce06763f114b4014552f062fd1f244328fe35b6ad