Helper Classes

Utility classes that are used by the product classes and by the Customizer.

assoc

Assoc Module to implement functions to read / write from so called "Assoc" data structures in Content Server.

Right now this module is used to tweak settings in XML-based transport packages that include Assoc structures inside some of the XML elements.

Classes:

Name	Description
Assoc	Class Assoc is used to handle Extended ECM Assoc data structures.

Assoc

Class Assoc is used to handle Extended ECM Assoc data structures.

Methods:

Name	Description
is_unicode_escaped	Determine if a string is unicode escaped.
escape_unicode	Escape / Encode a given string in Unicode.
unescape_unicode	Unescape / Decode a given string.
is_html_escaped	Check if an Assoc String is HTML escaped.
unescape_html	HTML unescape a string.
string_to_dict	Convert an Assoc string to a Python dict.
dict_to_string	Convert a Python dict to an Assoc string.
extract_substring	Extract a substring that is delimited by a start and stop sequence.
extract_assoc_string	Extract an Assoc from a string.

is_unicode_escaped classmethod

is_unicode_escaped(assoc_string: str) -> bool

Determine if a string is unicode escaped.

Parameters:

Name	Type	Description	Default
assoc_string	str	String with the Assoc data.	required

Returns:

Name	Type	Description
bool	bool	True if the string contains unicode escape sequences, False otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def is_unicode_escaped(cls, assoc_string: str) -> bool:
    """Determine if a string is unicode escaped.

    Args:
        assoc_string (str):
            String with the Assoc data.

    Returns:
        bool:
            True if the string contains unicode escape sequences, False otherwise.

    """

    pattern = r"\\u[0-9a-fA-F]{4}"
    matches = re.findall(pattern, assoc_string)

    return len(matches) > 0

escape_unicode classmethod

escape_unicode(assoc_string: str) -> bytes

Escape / Encode a given string in Unicode.

Parameters:

Name	Type	Description	Default
assoc_string	str	The source string to escape.	required

Returns:

Name	Type	Description
bytes	bytes	The escaped string as bytes.

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def escape_unicode(cls, assoc_string: str) -> bytes:
    """Escape / Encode a given string in Unicode.

    Args:
        assoc_string (str):
            The source string to escape.

    Returns:
        bytes:
            The escaped string as bytes.

    """

    encoded_string = assoc_string.encode("unicode_escape")  # .decode()

    return encoded_string

unescape_unicode classmethod

unescape_unicode(assoc_string: str) -> str

Unescape / Decode a given string.

Parameters:

Name	Type	Description	Default
assoc_string	str	The source string to unescape.	required

Returns:

Name	Type	Description
str	str	The unescaped string.

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def unescape_unicode(cls, assoc_string: str) -> str:
    """Unescape / Decode a given string.

    Args:
        assoc_string (str):
            The source string to unescape.

    Returns:
        str:
            The unescaped string.

    """
    try:
        decoded_string = bytes(assoc_string, "utf-8").decode("unicode_escape")
    except UnicodeDecodeError:
        return assoc_string

    return decoded_string

is_html_escaped classmethod

is_html_escaped(assoc_string: str) -> bool

Check if an Assoc String is HTML escaped.

Parameters:

Name	Type	Description	Default
assoc_string	str	The string to test for HTML escaping.	required

Returns:

Name	Type	Description
bool	bool	True if string is HTML escaped, False if not

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def is_html_escaped(cls, assoc_string: str) -> bool:
    """Check if an Assoc String is HTML escaped.

    Args:
        assoc_string (str):
            The string to test for HTML escaping.

    Returns:
        bool:
            True if string is HTML escaped, False if not

    """

    decoded_string = html.unescape(assoc_string)

    return assoc_string != decoded_string

unescape_html classmethod

unescape_html(assoc_string: str) -> str

HTML unescape a string.

Parameters:

Name	Type	Description	Default
assoc_string	str	The string to unescape.	required

Returns:

Name	Type	Description
str	str	The unescaped string.

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def unescape_html(cls, assoc_string: str) -> str:
    """HTML unescape a string.

    Args:
        assoc_string (str):
            The string to unescape.

    Returns:
        str:
            The unescaped string.

    """

    decoded_string = html.unescape(assoc_string)
    return decoded_string

string_to_dict classmethod

string_to_dict(assoc_string: str) -> dict

Convert an Assoc string to a Python dict.

Each comma-separated element of the Assoc will become a dict element.

Parameters:

Name	Type	Description	Default
assoc_string	str	The source Assoc string to convert.	required

Returns:

Name	Type	Description
dict	dict	Python dict with the Assoc elements

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def string_to_dict(cls, assoc_string: str) -> dict:
    """Convert an Assoc string to a Python dict.

    Each comma-separated element of the Assoc will become a dict element.

    Args:
        assoc_string (str):
            The source Assoc string to convert.

    Returns:
        dict: Python dict with the Assoc elements

    """

    if cls.is_html_escaped(assoc_string):
        assoc_string = cls.unescape_html(assoc_string)
    if cls.is_unicode_escaped(assoc_string):
        assoc_string = cls.unescape_unicode(assoc_string)

    # Split the string using regex pattern
    pieces = re.split(r",(?=(?:[^']*'[^']*')*[^']*$)", assoc_string)

    # Trim any leading/trailing spaces from each piece
    pieces = [piece.strip() for piece in pieces]

    # Split the last pieces from the assoc close tag
    last_piece = pieces[-1].split(">")[0]

    # Remove the first two and last pieces from the list
    # the first two are mostly "1" and "?"
    pieces = pieces[2:-1]

    # Insert the last pieces separately
    pieces.append(last_piece)

    assoc_dict: dict = {}

    for piece in pieces:
        name = piece.split("=")[0]
        if name[0] == "'":
            name = name[1:]
        if name[-1] == "'":
            name = name[:-1]
        value = piece.split("=")[1]
        if value[0] == "'":
            value = value[1:]
        if value[-1] == "'":
            value = value[:-1]
        assoc_dict[name] = value

    return assoc_dict

dict_to_string classmethod

dict_to_string(assoc_dict: dict) -> str

Convert a Python dict to an Assoc string.

Parameters:

Name	Type	Description	Default
assoc_dict	dict	The source dictionary to convert.	required

Returns:

Name	Type	Description
str	str	The resulting Assoc string.

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def dict_to_string(cls, assoc_dict: dict) -> str:
    """Convert a Python dict to an Assoc string.

    Args:
        assoc_dict (dict):
            The source dictionary to convert.

    Returns:
        str:
            The resulting Assoc string.

    """

    assoc_string: str = "A<1,?,"

    for item in assoc_dict.items():
        assoc_string += "\u0027" + item[0] + "\u0027"
        assoc_string += "="
        # Extended ECM's XML is a bit special in cases.
        # If the value is empty set (curly braces) it does
        # not put it in quotes. As Extended ECM is also very
        # picky about XML syntax we better produce it exactly like that.
        if item[1] == "{}":
            assoc_string += item[1] + ","
        else:
            assoc_string += "\u0027" + item[1] + "\u0027,"

    if assoc_dict.items():
        assoc_string = assoc_string[:-1]
    assoc_string += ">"
    return assoc_string

extract_substring classmethod

extract_substring(
    input_string: str,
    start_sequence: str,
    stop_sequence: str,
) -> str | None

Extract a substring that is delimited by a start and stop sequence.

Parameters:

Name	Type	Description	Default
input_string	str	Input string to search the delimited substring in.	required
start_sequence	str	Start sequence of characters.	required
stop_sequence	str	Stop sequence of characters.	required

Returns:

Type	Description
str | None	str | None: The delimited substring (including start and stop sequence) or None if not found.

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def extract_substring(
    cls,
    input_string: str,
    start_sequence: str,
    stop_sequence: str,
) -> str | None:
    """Extract a substring that is delimited by a start and stop sequence.

    Args:
        input_string (str):
            Input string to search the delimited substring in.
        start_sequence (str):
            Start sequence of characters.
        stop_sequence (str):
            Stop sequence of characters.

    Returns:
        str | None:
            The delimited substring (including start and stop sequence) or None if not found.

    """

    start_index = input_string.find(start_sequence)
    if start_index == -1:
        return None

    end_index = input_string.find(stop_sequence, start_index)
    if end_index == -1:
        return None

    end_index += len(stop_sequence)
    return input_string[start_index:end_index]

extract_assoc_string classmethod

extract_assoc_string(
    input_string: str, is_escaped: bool = False
) -> str

Extract an Assoc from a string.

The assoc is delimited by A< ... >.

Parameters:

Name	Type	Description	Default
input_string	str	Input string that includes the Assoc as a substring.	required
is_escaped	bool	Whether or not the input string includes the assoc escaped or not. Defaults to False.	False

Returns:

Name	Type	Description
str	str	The assoc string.

Source code in packages/pyxecm/src/pyxecm/helper/assoc.py

@classmethod
def extract_assoc_string(cls, input_string: str, is_escaped: bool = False) -> str:
    """Extract an Assoc from a string.

    The assoc is delimited by A< ... >.

    Args:
        input_string (str):
            Input string that includes the Assoc as a substring.
        is_escaped (bool, optional):
            Whether or not the input string includes the
            assoc escaped or not. Defaults to False.

    Returns:
        str:
            The assoc string.

    """

    if is_escaped:
        assoc_string = cls.extract_substring(input_string, "A&lt;", "&gt;")
    else:
        assoc_string = cls.extract_substring(input_string, "A<", ">")
    return assoc_string

data

Data Module leveraging Pandas to manipulate data sets read for bulk generation of Content Server items.

See: https://pandas.pydata.org

This code implements a class called "Data" which is a wrapper to Pandas data frame.

Classes:

Name	Description
Data	Used to handle big data sets that have a tabular structure.

Data

Data(
    init_data: DataFrame | list = None,
    columns: list[str] | None = None,
    dtypes: dict[str, type] | None = None,
    index_columns: str | list[str] | None = None,
    logger: Logger = default_logger,
)

Used to handle big data sets that have a tabular structure.

Initialize the Data object.

Parameters:

Name	Type	Description	Default
init_data	DataFrame | list	Data to initialize the data frame. Can either be another data frame or Data object (that gets copied), a list of dictionaries or a single dictionary (which becomes one row). Defaults to None.	None
columns	list[str] | None	The list of column names to use if init_data is None, a list or a dictionary. Defaults to None.	None
dtypes	dict[str, type] | None	A dictionary defining the data types for specific columns. The keys are the column names and the values are the desired data types. Defaults to None. In this case pandas will infer the data types automatically.	None
index_columns	str | list[str] | None	The name of the column that should become the index of the data frame. Defaults to None. Also a list of column names can be provided to create a multi-index.	None
logger	Logger	Pass a special logging object. This is optional. If not provided, the default logger is used.	default_logger

Methods:

Name	Description
read_sql	Load a Data object directly from a SQL table/query.
lock	Return the threading lock object.
get_data_frame	Get the Pandas data frame object.
set_data_frame	Set the Pandas data frame object.
get_columns	Get the list of column names of the data frame.
print_info	Log information about the data frame.
append	Append additional data to the data frame.
apply_schema	Force the internal DataFrame to match the defined schema and index settings.
merge	Merge the current DataFrame (_df) with another DataFrame.
strip	Strip leading and trailing spaces from specified columns in a data frame.
load_json_data	Load JSON data into a Pandas data frame.
save_json_data	Save JSON data from data frame to file.
load_excel_data	Load Excel (xlsx) data into Pandas data frame.
save_excel_data	Save the data frame to an Excel file, with robust error handling and logging.
load_csv_data	Load CSV (Comma separated values) data into data frame.
load_xml_data	Load XML data into a Pandas data frame.
load_directory	Load directory structure into Pandas data frame.
load_xml_directory	Load XML files from a directory structure into Pandas data frame.
load_web_links	Get all linked file URLs on a given web page (url) that are following a given pattern.
load_web	Traverse years and bulletin types to collect all bulletin URLs.
partitionate	Partition a data frame into equally sized partitions.
partitionate_by_column	Partition a data frame based on equal values in a specified column.
deduplicate	Remove duplicate rows that have all fields in unique_fields in common.
sort	Sort the data frame based on one or multiple fields.
flatten	Flatten a sub-dictionary by copying selected fields to the parent dictionary.
explode_and_flatten	Explode a substructure in the Pandas data frame.
drop_columns	Drop selected columns from the Pandas data frame.
keep_columns	Keep only selected columns in the data frame. Drop the rest.
rename_column	Rename a data frame column.
is_dict_column	Safely checks if a column predominantly contains dictionary-like objects.
is_list_column	Safely checks if a column predominantly contains list-like objects.
is_string_column	Determine if a Pandas series contains only string values, ignoring NaN values.
cleanse	Cleanse data with regular expressions and upper/lower case conversions.
filter	Filter the data frame based on (multiple) conditions.
fill_na_in_column	Replace NA values in a column with a defined new default value.
fill_forward	Fill the missing cells appropriately by carrying forward the values from the previous rows where necessary.
lookup_value	Lookup row(s) that includes a lookup value in the value of a given column.
set_value	Set the value in the data frame based on a condition.
add_column	Add additional column to the data frame.
convert_to_lists	Intelligently convert string values to list values, in defined data frame columns.
add_column_concat	Add a column as a concatenation of the values of multiple source columns.
add_column_list	Add a column with list objects.
add_column_table	Add a column with tabular objects (list of dictionaries).
drop_row	Drop a single row from the DataFrame by its index.
drop_rows	Drop rows from the DataFrame based on a boolean mask.
get_match_mask	Find rows in this Data object that match another Data object or DataFrame.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def __init__(
    self,
    init_data: pd.DataFrame | list = None,
    columns: list[str] | None = None,
    dtypes: dict[str, type] | None = None,
    index_columns: str | list[str] | None = None,
    logger: logging.Logger = default_logger,
) -> None:
    """Initialize the Data object.

    Args:
        init_data (pd.DataFrame | list, optional):
            Data to initialize the data frame. Can either be
            another data frame or Data object (that gets copied), a list of dictionaries
            or a single dictionary (which becomes one row). Defaults to None.
        columns (list[str] | None, optional):
            The list of column names to use if init_data is None, a list or a dictionary.
            Defaults to None.
        dtypes (dict[str, type] | None, optional):
            A dictionary defining the data types for specific columns.
            The keys are the column names and the values are the desired data types.
            Defaults to None. In this case pandas will infer the data types automatically.
        index_columns (str | list[str] | None, optional):
            The name of the column that should become the index
            of the data frame. Defaults to None.
            Also a list of column names can be provided to create
            a multi-index.
        logger (logging.Logger, optional):
            Pass a special logging object. This is optional. If not provided,
            the default logger is used.

    """

    if logger != default_logger:
        self.logger = logger.getChild("data")
        for logfilter in logger.filters:
            self.logger.addFilter(logfilter)

    self._lock: threading.Lock = threading.Lock()

    self._schema = dtypes

    if init_data is None:
        self._df = pd.DataFrame(columns=columns)
    elif isinstance(init_data, pd.DataFrame):
        self._df = init_data.copy()
    elif isinstance(init_data, Data):
        self._df = init_data.get_data_frame().copy()
    elif isinstance(init_data, dict):
        # We don't want Pandas to apply any "magic" here, so we wrap
        # the dict in a list to ensure it is treated as a single row
        self._df = pd.DataFrame(data=[init_data], columns=columns)
    elif isinstance(init_data, list):
        self._df = pd.DataFrame(data=init_data, columns=columns)
    else:
        error_message = "Illegal initialization data for 'Data' class!"
        raise TypeError(error_message)

    # Apply dtypes if given
    if self._schema:
        for col, dtype in self._schema.items():
            if col in self._df.columns:
                if dtype.startswith("datetime64") and "UTC" in dtype:
                    self._df[col] = pd.to_datetime(self._df[col], utc=True)
                elif dtype.startswith("datetime64"):
                    self._df[col] = pd.to_datetime(self._df[col])
                else:
                    self._df[col] = self._df[col].astype(dtype)

    # set index if specified
    if index_columns:
        # convert single string to list for uniformity
        if isinstance(index_columns, str):
            self._index_columns = [index_columns]
        else:
            self._index_columns = index_columns

        # check that all index columns exist in the DataFrame
        missing_cols = [col for col in self._index_columns if col not in self._df.columns]
        if missing_cols:
            msg = f"Cannot set index. Missing columns in DataFrame: {missing_cols}"
            raise ValueError(msg)

        # set the index
        self._df.set_index(self._index_columns, inplace=True, drop=False)
    else:
        self._index_columns = None

read_sql classmethod

read_sql(
    sql: str,
    con: any,
    columns: list[str] | None = None,
    dtypes: dict[str, type | str] | None = None,
    index_columns: str | list[str] | None = None,
    **kwargs: dict,
) -> Data

Load a Data object directly from a SQL table/query.

Parameters:

Name	Type	Description	Default
sql	str	SQL query string or SQLAlchemy Table.	required
con	any	Database connection or engine.	required
columns	list[str] | None	List of columns to load. Defaults to None. This is passed to pd.read_sql which only uses it if sql is a table name.	None
dtypes	dict[str, type | str] | None	Column dtype dictionary (NumPy/Pandas types or strings). Defaults to None.	None
index_columns	str | list[str] | None	The name of the column that should become the index of the data frame. Defaults to None. Also a list of column names can be provided to create a multi-index.	None
**kwargs	dict	Extra kwargs passed to pd.read_sql.	{}

Returns:

Name	Type	Description
Data	Data	A new Data object with the result of the SQL query.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

@classmethod
def read_sql(
    cls,
    sql: str,
    con: any,
    columns: list[str] | None = None,
    dtypes: dict[str, type | str] | None = None,
    index_columns: str | list[str] | None = None,
    **kwargs: dict,
) -> Data:
    """Load a Data object directly from a SQL table/query.

    Args:
        sql (str):
            SQL query string or SQLAlchemy Table.
        con (any):
            Database connection or engine.
        columns (list[str] | None, optional):
            List of columns to load. Defaults to None. This is passed to `pd.read_sql`
            which only uses it if `sql` is a table name.
        dtypes (dict[str, type | str] | None, optional):
            Column dtype dictionary (NumPy/Pandas types or strings). Defaults to None.
        index_columns (str | list[str] | None, optional):
            The name of the column that should become the index
            of the data frame. Defaults to None.
            Also a list of column names can be provided to create
            a multi-index.
        **kwargs (dict):
            Extra kwargs passed to pd.read_sql.

    Returns:
        Data:
            A new Data object with the result of the SQL query.

    """
    df = pd.read_sql(sql, con=con, columns=columns, dtype=dtypes, **kwargs)
    return cls(init_data=df, dtypes=dtypes, index_columns=index_columns)

lock

lock() -> Lock

Return the threading lock object.

Returns:

Type	Description
Lock	threading.Lock: The threading lock object.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def lock(self) -> threading.Lock:
    """Return the threading lock object.

    Returns:
        threading.Lock: The threading lock object.

    """

    return self._lock

get_data_frame

get_data_frame() -> DataFrame

Get the Pandas data frame object.

Returns:

Type	Description
DataFrame	pd.DataFrame: The Pandas data frame object.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def get_data_frame(self) -> pd.DataFrame:
    """Get the Pandas data frame object.

    Returns:
        pd.DataFrame: The Pandas data frame object.

    """

    return self._df

set_data_frame

set_data_frame(df: DataFrame) -> None

Set the Pandas data frame object.

Parameters:

Name	Type	Description	Default
df	DataFrame	The new Pandas data frame object.	required

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def set_data_frame(self, df: pd.DataFrame) -> None:
    """Set the Pandas data frame object.

    Args:
        df (pd.DataFrame): The new Pandas data frame object.

    """

    self._df = df

get_columns

get_columns() -> list | None

Get the list of column names of the data frame.

Returns:

Type	Description
list | None	list | None: The list of column names in the data frame.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def get_columns(self) -> list | None:
    """Get the list of column names of the data frame.

    Returns:
        list | None:
            The list of column names in the data frame.

    """

    if self._df is None:
        return None

    return self._df.columns

print_info

print_info(
    show_size: bool = True,
    show_info: bool = False,
    show_columns: bool = False,
    show_first: bool = False,
    show_last: bool = False,
    show_sample: bool = False,
    show_statistics: bool = False,
    row_num: int = 10,
) -> None

Log information about the data frame.

Parameters:

Name	Type	Description	Default
show_size	bool	Show size of data frame. Defaults to True.	True
show_info	bool	Show information for data frame. Defaults to False.	False
show_columns	bool	Show columns of data frame. Defaults to False.	False
show_first	bool	Show first N items. Defaults to False. N is defined by the row_num parameter.	False
show_last	bool	Show last N items. Defaults to False. N is defined by the row_num parameter.	False
show_sample	bool	Show N sample items. Defaults to False. N is defined by the row_num parameter.	False
show_statistics	bool	Show data frame statistics. Defaults to False.	False
row_num	int	Used as the number of rows printed using show_first, show_last, show_sample. Default is 10.	10

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def print_info(
    self,
    show_size: bool = True,
    show_info: bool = False,
    show_columns: bool = False,
    show_first: bool = False,
    show_last: bool = False,
    show_sample: bool = False,
    show_statistics: bool = False,
    row_num: int = 10,
) -> None:
    """Log information about the data frame.

    Args:
        show_size (bool, optional):
            Show size of data frame. Defaults to True.
        show_info (bool, optional):
            Show information for data frame. Defaults to False.
        show_columns (bool, optional):
            Show columns of data frame. Defaults to False.
        show_first (bool, optional):
            Show first N items. Defaults to False. N is defined
            by the row_num parameter.
        show_last (bool, optional):
            Show last N items. Defaults to False. N is defined
            by the row_num parameter.
        show_sample (bool, optional):
            Show N sample items. Defaults to False. N is defined
            by the row_num parameter.
        show_statistics (bool, optional):
            Show data frame statistics. Defaults to False.
        row_num (int, optional):
            Used as the number of rows printed using show_first,
            show_last, show_sample. Default is 10.

    """

    if self._df is None:
        self.logger.warning("Data frame is not initialized!")
        return

    if show_size:
        self.logger.info(
            "Data frame has %s row(s) and %s column(s)",
            self._df.shape[0],
            self._df.shape[1],
        )

    if show_info:
        # df.info() can not easily be embedded into a string
        self._df.info()

    if show_columns:
        self.logger.info("Columns:\n%s", self._df.columns)
        self.logger.info(
            "Columns with number of NaN values:\n%s",
            self._df.isna().sum(),
        )
        self.logger.info(
            "Columns with number of non-NaN values:\n%s",
            self._df.notna().sum(),
        )

    if show_first:
        # the default for head is n = 5:
        self.logger.info("First %s rows:\n%s", str(row_num), self._df.head(row_num))

    if show_last:
        # the default for tail is n = 5:
        self.logger.info("Last %s rows:\n%s", str(row_num), self._df.tail(row_num))

    if show_sample:
        # the default for sample is n = 1:
        self.logger.info(
            "%s Sample rows:\n%s",
            str(row_num),
            self._df.sample(n=row_num),
        )

    if show_statistics:
        self.logger.info(
            "Description of statistics for data frame:\n%s",
            self._df.describe(),
        )
        self.logger.info(
            "Description of statistics for data frame (transformed):\n%s",
            self._df.describe().T,
        )
        self.logger.info(
            "Description of statistics for data frame (objects):\n%s",
            self._df.describe(include="object"),
        )

append

append(
    add_data: DataFrame | list | dict | Data,
    enforce_schema: bool = False,
) -> bool

Append additional data to the data frame.

Note: This method is not thread-safe; locking must be handled by the caller.

Parameters:

Name	Type	Description	Default
add_data	DataFrame | list | dict | Data	Data to append (DataFrame, list of dicts, dict, or Data object).	required
enforce_schema	bool	If True, aligns columns to self._schema and enforces dtypes. Defaults to False to maintain backwards compatibility.	False

Returns:

Name	Type	Description
bool	bool	True if successful, False otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def append(self, add_data: pd.DataFrame | list | dict | Data, enforce_schema: bool = False) -> bool:
    """Append additional data to the data frame.

    Note: This method is not thread-safe; locking must be handled by the caller.

    Args:
        add_data (pd.DataFrame | list | dict | Data):
            Data to append (DataFrame, list of dicts, dict, or Data object).
        enforce_schema (bool, optional):
            If True, aligns columns to self._schema and enforces dtypes.
            Defaults to False to maintain backwards compatibility.

    Returns:
        bool:
            True if successful, False otherwise.

    """
    try:
        # Prepare construction parameters if schema is requested
        target_cols: list[str] | None = list(self._schema.keys()) if (enforce_schema and self._schema) else None

        # 1. Normalize input to a pandas DataFrame
        if isinstance(add_data, Data):
            new_df = add_data.get_data_frame()
        elif isinstance(add_data, pd.DataFrame):
            new_df = add_data
        elif isinstance(add_data, (list, dict)):
            if not add_data:
                return True

            # Wrap single dict in list to ensure it is treated as one row
            add_data = [add_data] if isinstance(add_data, dict) else add_data

            # Early Enforcement: Pass columns to the constructor
            new_df = pd.DataFrame(data=add_data, columns=target_cols)
        else:
            self.logger.error("Illegal data type -> '%s' for append!", type(add_data))
            return False

        # 2. Apply Schema & Index Logic
        if enforce_schema and self._schema is not None:
            # Always reindex/astype if enforcing schema to ensure 100% alignment
            new_df = new_df.reindex(columns=self._schema.keys())

            for col, dtype in self._schema.items():
                if dtype.startswith("datetime64"):
                    if "UTC" in dtype:
                        new_df[col] = pd.to_datetime(new_df[col], utc=True)
                    else:
                        new_df[col] = pd.to_datetime(new_df[col])
                else:
                    new_df[col] = new_df[col].astype(dtype)

            if self._index_columns is not None:
                new_df.set_index(self._index_columns, inplace=True, drop=False)

        # 3. Combine Data
        if self._df is None or self._df.empty:
            # If enforcing schema, we take the processed new_df
            # If not, we just take the data as-is
            self._df = new_df.copy()
        else:
            # enforce_schema=False uses ignore_index=True (standard append behavior)
            # enforce_schema=True uses ignore_index=False (preserves the index we just set)
            ignore_idx = not enforce_schema
            self._df = pd.concat([self._df, new_df], ignore_index=ignore_idx)

    except Exception as e:
        self.logger.error("Append failed; error -> %s", str(e))
        return False
    else:
        return True

apply_schema

apply_schema(
    apply_dtypes: bool = True, apply_index: bool = True
) -> None

Force the internal DataFrame to match the defined schema and index settings.

Parameters:

Name	Type	Description	Default
apply_dtypes	bool	If True, reindexes columns and casts them to the dtypes defined in self._schema. Defaults to True.	True
apply_index	bool	If True, re-establishes the index based on self._index_columns. Defaults to True.	True

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def apply_schema(self, apply_dtypes: bool = True, apply_index: bool = True) -> None:
    """Force the internal DataFrame to match the defined schema and index settings.

    Args:
        apply_dtypes (bool, optional):
            If True, reindexes columns and casts them to the dtypes
            defined in self._schema. Defaults to True.
        apply_index (bool, optional):
            If True, re-establishes the index based on self._index_columns.
            Defaults to True.

    """

    if self._df is None:
        self.logger.warning("Cannot apply schema on uninitialized data object.")
        return

    try:
        # 1. Handle Dtypes and Column Alignment
        if apply_dtypes and self._schema:
            # Reindex ensures order and presence. We cast per-column so we can
            # treat datetime dtypes via pd.to_datetime (timezone metadata such as
            # "datetime64[s,UTC]" is not accepted by astype()).
            self._df = self._df.reindex(columns=self._schema.keys())

            for col, dtype in self._schema.items():
                if dtype.startswith("datetime64") and "UTC" in dtype:
                    self._df[col] = pd.to_datetime(self._df[col], utc=True)
                elif dtype.startswith("datetime64"):
                    self._df[col] = pd.to_datetime(self._df[col])
                else:
                    self._df[col] = self._df[col].astype(dtype)

            self.logger.debug("Applied dtypes and column alignment.")

        # 2. Handle Index Alignment
        if apply_index and self._index_columns:
            # We check if the index is already correctly named to avoid
            # unnecessary re-indexing overhead.
            current_index_names = self._df.index.names
            target_index_names = (
                [self._index_columns] if isinstance(self._index_columns, str) else self._index_columns
            )

            if current_index_names != target_index_names:
                self._df.set_index(self._index_columns, inplace=True, drop=False)
                self.logger.debug("Applied index settings -> %s", self._index_columns)

    except Exception as e:
        self.logger.error("Failed to apply schema: %s", str(e))
        raise

merge

merge(
    merge_data: DataFrame,
    on: str | list[str] | None = None,
    how: str = "inner",
    left_on: str | list[str] | None = None,
    right_on: str | list[str] | None = None,
    left_index: bool = False,
    right_index: bool = False,
    suffixes: tuple[str, str] = ("_x", "_y"),
    indicator: bool = False,
    validate: str | None = None,
) -> DataFrame | None

Merge the current DataFrame (_df) with another DataFrame.

Parameters:

Name	Type	Description	Default
merge_data	DataFrame | Data	The DataFrame to merge with.	required
on	str | list[str] | None	Column(s) to merge on. Defaults to None.	None
how	str	Type of merge ('inner', 'outer', 'left', 'right', 'cross'). Defaults to 'inner'.	'inner'
left_on	str | list[str] | None	Column(s) from self._df to merge on. Defaults to None.	None
right_on	str | list[str] | None	Column(s) from other DataFrame to merge on. Defaults to None.	None
left_index	bool	Whether to merge on the index of self._df. Defaults to False.	False
right_index	bool	Whether to merge on the index of other. Defaults to False.	False
suffixes	tuple[str, str]	Suffixes for overlapping column names. Defaults to ('_x', '_y').	('_x', '_y')
indicator	bool	If True, adds a column showing the merge source. Defaults to False.	False
validate	str | None	If provided, checks merge integrity ('one_to_one', 'one_to_many', 'many_to_one', 'many_to_many'). Defaults to None.	None

Returns:

Type	Description
DataFrame | None	pd.DataFrame | None: The merged DataFrame or None in case of an error (missing columns for merging or an invalid merge operation, e.g. if the validate check fails).

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def merge(
    self,
    merge_data: pd.DataFrame,
    on: str | list[str] | None = None,
    how: str = "inner",
    left_on: str | list[str] | None = None,
    right_on: str | list[str] | None = None,
    left_index: bool = False,
    right_index: bool = False,
    suffixes: tuple[str, str] = ("_x", "_y"),
    indicator: bool = False,
    validate: str | None = None,
) -> pd.DataFrame | None:
    """Merge the current DataFrame (_df) with another DataFrame.

    Args:
        merge_data (pd.DataFrame | Data):
            The DataFrame to merge with.
        on (str | list[str] | None, optional):
            Column(s) to merge on. Defaults to None.
        how (str, optional):
            Type of merge ('inner', 'outer', 'left', 'right', 'cross'). Defaults to 'inner'.
        left_on (str | list[str] | None, optional):
            Column(s) from self._df to merge on. Defaults to None.
        right_on (str | list[str] | None, optional):
            Column(s) from other DataFrame to merge on. Defaults to None.
        left_index (bool, optional):
            Whether to merge on the index of self._df. Defaults to False.
        right_index (bool, optional):
            Whether to merge on the index of other. Defaults to False.
        suffixes (tuple[str, str], optional):
            Suffixes for overlapping column names. Defaults to ('_x', '_y').
        indicator (bool, optional):
            If True, adds a column showing the merge source. Defaults to False.
        validate (str | None, optional):
            If provided, checks merge integrity
            ('one_to_one', 'one_to_many', 'many_to_one', 'many_to_many'). Defaults to None.

    Returns:
        pd.DataFrame | None:
            The merged DataFrame or None in case of an error (missing columns for
            merging or an invalid merge operation, e.g. if the `validate` check fails).

    """

    if self._df is None or self._df.empty:
        self._df = merge_data

    if isinstance(merge_data, Data):
        merge_data = merge_data.get_data_frame()  # Extract DataFrame from Data instance

    try:
        return self._df.merge(
            merge_data,
            how=how,
            on=on,
            left_on=left_on,
            right_on=right_on,
            left_index=left_index,
            right_index=right_index,
            suffixes=suffixes,
            indicator=indicator,
            validate=validate,
        )
    except KeyError:
        self.logger.error("Column(s) not found for merging!")
    except ValueError:
        self.logger.error("Invalid merge operation!")

    return None

strip

strip(
    columns: list | None = None, inplace: bool = True
) -> DataFrame

Strip leading and trailing spaces from specified columns in a data frame.

Parameters:

Name	Type	Description	Default
columns	list | None	The list of column names to strip. If None, it strips leading and trailing spaces from all string columns. Defaults to None.	None
inplace	bool	If True, the data modification is done in place, i.e. modifying the existing data frame of the object. If False, the data frame is copied and the copy is modified and returned. Defaults to True.	True

Returns:

Type	Description
DataFrame	pd.DataFrame: The modified data frame with stripped columns.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def strip(self, columns: list | None = None, inplace: bool = True) -> pd.DataFrame:
    """Strip leading and trailing spaces from specified columns in a data frame.

    Args:
        columns (list | None, optional):
            The list of column names to strip. If None, it strips
            leading and trailing spaces from _all_ string columns. Defaults to None.
        inplace (bool, optional):
            If True, the data modification is done in place, i.e.
            modifying the existing data frame of the object.
            If False, the data frame is copied and the copy is modified
            and returned. Defaults to True.

    Returns:
        pd.DataFrame:
            The modified data frame with stripped columns.

    """

    df = self._df.copy() if not inplace else self._df

    if columns is None:
        # Strip spaces from all string columns
        df = df.apply(lambda x: x.str.strip() if x.dtype == "object" else x)
    else:
        # Strip spaces from specified columns
        for col in columns:
            if col in df.columns and df[col].dtype == "object":  # Check if the column exists and is of string type
                df[col] = df[col].str.strip()

    if inplace:
        self._df = df

    return df

load_json_data

load_json_data(
    json_path: str,
    convert_dates: bool = False,
    index_column: str | None = None,
    compression: str | None = None,
    replace: bool = False,
) -> bool

Load JSON data into a Pandas data frame.

Without a schema (no dtypes given when the Data object was created), pandas infers the data types.

With a schema, the JSON file is loaded with the schema (e.g. data saved with save_json_data()): pandas does not infer the data types (it would e.g. turn the string "1e3" into the number 1000.0), and the schema is applied after loading (see apply_schema()): the columns of the schema with their data types (also datetimes, if saved in ISO format) and the index (index_columns). If the schema cannot be applied, the data frame is left unchanged.

Parameters:

Name	Type	Description	Default
json_path	str	The path to the JSON file.	required
convert_dates	bool	Defines whether or not dates should be converted. The default is False = dates are NOT converted. With a schema, the datetime columns of the schema are converted anyway.	False
index_column	str | None	The Name of the column (i.e. JSON data field) that should become the index in the loaded data frame. Defaults to None. Not to be combined with a schema - its index (index_columns) is applied.	None
compression	str | None	The compression of the JSON file that is to be decompressed: * gzip (.gz) * bz2 (.bz2) * zip (.zip) * xz (.xz) The value for compression should not include the dot. Default is None = no compression.	None
replace	bool	If True, the loaded data replaces the content of the data frame. If False (default), it is appended to the data frame (with a new row index).	False

Returns:

Name	Type	Description
bool	bool	False in case an error occurred, True otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_json_data(
    self,
    json_path: str,
    convert_dates: bool = False,
    index_column: str | None = None,
    compression: str | None = None,
    replace: bool = False,
) -> bool:
    """Load JSON data into a Pandas data frame.

    Without a schema (no `dtypes` given when the Data object was created), pandas infers the
    data types.

    With a schema, the JSON file is loaded with the schema (e.g. data saved with save_json_data()):
    pandas does not infer the data types (it would e.g. turn the string "1e3" into the number
    1000.0), and the schema is applied after loading (see apply_schema()): the columns of the
    schema with their data types (also datetimes, if saved in ISO format) and the index
    (`index_columns`). If the schema cannot be applied, the data frame is left unchanged.

    Args:
        json_path (str):
            The path to the JSON file.
        convert_dates (bool, optional):
            Defines whether or not dates should be converted.
            The default is False = dates are NOT converted. With a schema, the datetime
            columns of the schema are converted anyway.
        index_column (str | None, optional):
            The Name of the column (i.e. JSON data field) that should
            become the index in the loaded data frame. Defaults to None.
            Not to be combined with a schema - its index (`index_columns`) is applied.
        compression (str | None, optional):
            The compression of the JSON file that is to be decompressed:
            * gzip (.gz)
            * bz2 (.bz2)
            * zip (.zip)
            * xz (.xz)
            The value for compression should not include the dot.
            Default is None = no compression.
        replace (bool, optional):
            If True, the loaded data replaces the content of the data frame. If False (default),
            it is appended to the data frame (with a new row index).

    Returns:
        bool: False in case an error occurred, True otherwise.

    """

    if not json_path:
        self.logger.error(
            "You have not specified a JSON path!",
        )
        return False

    # If compression is enabled the file path should have
    # the matching file name extension:
    if compression:
        compression = compression.lstrip(".")  # remove a dot prefix if present
        suffix = "." + compression if compression != "gzip" else ".gz"
        if not json_path.endswith(suffix):
            json_path += suffix

    if not os.path.exists(json_path):
        self.logger.error(
            "Missing JSON file - you have not specified a valid path -> '%s'.",
            json_path,
        )
        return False

    # Load data from JSON file
    try:
        df = pd.read_json(
            path_or_buf=json_path,
            convert_dates=convert_dates,
            compression=compression,
            # With a schema the data types are defined by it - pandas must not infer them:
            dtype=not self._schema,
        )

        if index_column and index_column not in df.columns:
            self.logger.error(
                "Specified index column -> '%s' not found in the JSON data.",
                index_column,
            )
            return False

        if index_column:
            df = df.set_index(keys=index_column)
        previous_df = self._df
        if replace or self._df is None:
            self._df = df
        else:
            self._df = pd.concat([self._df, df], ignore_index=True)
        if self._schema:
            try:
                self.apply_schema()
            except Exception:
                # apply_schema() has logged the error - keep the data frame as it was:
                self._df = previous_df
                raise
        self.logger.info(
            "After loading JSON file -> '%s', the data frame has %s row(s) and %s column(s)",
            json_path,
            self._df.shape[0],
            self._df.shape[1],
        )
    except FileNotFoundError:
        self.logger.error(
            "JSON file -> '%s' not found. Please check the file path.",
            json_path,
        )
        return False
    except PermissionError:
        self.logger.error(
            "Missing permission to access the JSON file -> '%s'.",
            json_path,
        )
        return False
    except OSError:
        self.logger.error("An I/O error occurred!")
        return False
    except json.JSONDecodeError:
        self.logger.error(
            "Unable to decode JSON file -> '%s'",
            json_path,
        )
        return False
    except ValueError:
        self.logger.error("Invalid JSON input -> %s", json_path)
        return False
    except AttributeError:
        self.logger.error("Unexpected JSON data structure in file -> %s", json_path)
        return False
    except TypeError:
        self.logger.error("Unexpected JSON data type in file -> %s", json_path)
        return False
    except KeyError:
        self.logger.error("Missing key in JSON data in file -> %s", json_path)
        return False

    return True

save_json_data

save_json_data(
    json_path: str,
    orient: str = "records",
    preserve_index: bool = False,
    index_column: str = "index",
    compression: str | None = None,
    indent: int | None = 2,
) -> bool

Save JSON data from data frame to file.

Datetimes are saved in ISO format (with millisecond precision). A data frame saved with orient="records" is loaded again with its data types by load_json_data() of a Data object with the same schema.

Parameters:

Name	Type	Description	Default
json_path	str	The path to where the JSON file should be saved.	required
orient	str	The structure of the JSON. Possible values: * "records" (this is the default) * "columns" * "index" * "table" * "split"	'records'
preserve_index	bool	Defines if the index column of the data frame should be exported as well. The default is False (index is not exported).	False
index_column	str	The name of the column that is created to preserve the index of the data frame. This is only used if preserve_index is True and orient is not one of "columns", "index", "table" or "split". The default is "index".	'index'
compression	str | None	Apply a compression: * gzip (.gz) * bz2 (.bz2) * zip (.zip) * xz (.xz) Default is None = no compression.	None
indent	int | None	The number of spaces to indent the JSON. None writes compact JSON (smaller files). Defaults to 2.	2

Returns:

Name	Type	Description
bool	bool	False in case an error occurred, True otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def save_json_data(
    self,
    json_path: str,
    orient: str = "records",
    preserve_index: bool = False,
    index_column: str = "index",
    compression: str | None = None,
    indent: int | None = 2,
) -> bool:
    """Save JSON data from data frame to file.

    Datetimes are saved in ISO format (with millisecond precision). A data frame saved with
    `orient="records"` is loaded again with its data types by load_json_data() of a Data object
    with the same schema.

    Args:
        json_path (str):
            The path to where the JSON file should be saved.
        orient (str, optional):
            The structure of the JSON. Possible values:
            * "records" (this is the default)
            * "columns"
            * "index"
            * "table"
            * "split"
        preserve_index (bool, optional):
            Defines if the index column of the data frame should be exported as well.
            The default is False (index is not exported).
        index_column (str, optional):
            The name of the column that is created to preserve the index of the data frame.
            This is only used if preserve_index is True and orient is not one of
            "columns", "index", "table" or "split". The default is "index".
        compression (str | None, optional):
            Apply a compression:
            * gzip (.gz)
            * bz2 (.bz2)
            * zip (.zip)
            * xz (.xz)
            Default is None = no compression.
        indent (int | None, optional):
            The number of spaces to indent the JSON. None writes compact JSON (smaller files).
            Defaults to 2.

    Returns:
        bool:
            False in case an error occurred, True otherwise.

    """

    if not json_path:
        self.logger.error(
            "You have not specified a JSON path!",
        )
        return False

    # If compression is enabled the file path should have
    # the matching file name extension:
    if compression:
        suffix = "." + compression if compression != "gzip" else ".gz"
        if not json_path.endswith(suffix):
            json_path += suffix

    # Save data to JSON file
    try:
        if self._df is not None:
            # A file name without directory is saved in the current directory:
            json_directory = os.path.dirname(json_path)
            if json_directory:
                os.makedirs(json_directory, exist_ok=True)

            # index parameter is only allowed if orient has one of the following values:
            if orient in ("columns", "index", "table", "split"):
                self._df.to_json(
                    path_or_buf=json_path,
                    index=preserve_index,
                    orient=orient,
                    indent=indent,
                    compression=compression,
                    date_format="iso",
                )
            # In this case we cannot use the index parameter as this would give this error:
            # Value Error -> 'index=True' is only valid when 'orient' is 'split', 'table', 'index', or 'columns'
            # So we create a new column that preserves the original row IDs from the index. The nasme

            elif preserve_index:
                df_with_index = self._df.reset_index(
                    names=index_column,
                    inplace=False,
                )
                df_with_index.to_json(
                    path_or_buf=json_path,
                    orient=orient,
                    indent=indent,
                    compression=compression,
                    date_format="iso",
                )
            else:
                self._df.to_json(
                    path_or_buf=json_path,
                    orient=orient,
                    indent=indent,
                    compression=compression,
                    date_format="iso",
                )
        else:
            self.logger.warning(
                "Data frame is empty. Cannot write it to JSON file -> '%s'.",
                json_path,
            )
            return False
    except FileNotFoundError:
        self.logger.error(
            "File -> '%s' not found. Please check the file path.",
            json_path,
        )
        return False
    except PermissionError:
        self.logger.error(
            "Permission denied to access the file -> '%s'.",
            json_path,
        )
        return False
    except OSError:
        self.logger.error("An I/O error occurred accessing file -> %s", json_path)
        return False
    except ValueError:
        self.logger.error("Value error!")
        return False

    return True

load_excel_data

load_excel_data(
    xlsx_path: str,
    sheet_names: str | list | None = 0,
    usecols: str | list | None = None,
    skip_rows: int | None = None,
    header: int | None = 0,
    names: list | None = None,
    na_values: list | None = None,
) -> bool

Load Excel (xlsx) data into Pandas data frame.

Supports xls, xlsx, xlsm, xlsb, odf, ods and odt file extensions read from a local filesystem or URL. Supports an option to read a single sheet or a list of sheets.

Parameters:

Name	Type	Description	Default
xlsx_path	str	The path to the Excel file to load.	required
sheet_names	str | list | int | None	Name or Index of the sheet in the Excel workbook to load. If 'None' then all sheets will be loaded. If 0 then first sheet in workbook will be loaded (this is the Default). If string then this is interpreted as the name of the sheet to load. If a list is passed, this can be a list of index values (int) or a list of strings with the sheet names to load. If multiple sheets are loaded they are concatenated into one data frame.	0
usecols	list | str | None	A list of columns to load, specified by general column names in Excel, e.g. usecols='B:D', usecols=['A', 'C', 'F']. Defaults to None (all columns).	None
skip_rows	int | None	Number of rows to skip on top of the sheet (e.g. to not read headlines). Defaults to None.	None
header	int | None	Excel Row (0-indexed) to use for the column labels of the parsed data frame. If file contains no header row, then you should explicitly pass header=None. Default is 0.	0
names	list | None	A list of column names to use. Default is None.	None
na_values	list | None	A list of values in the Excel that should become the Pandas NA value. Defaults to None.	None

Returns:

Name	Type	Description
bool	bool	False in case an error occurred, True otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_excel_data(
    self,
    xlsx_path: str,
    sheet_names: str | list | None = 0,
    usecols: str | list | None = None,
    skip_rows: int | None = None,
    header: int | None = 0,
    names: list | None = None,
    na_values: list | None = None,
) -> bool:
    """Load Excel (xlsx) data into Pandas data frame.

    Supports xls, xlsx, xlsm, xlsb, odf, ods and odt file extensions
    read from a local filesystem or URL. Supports an option to read a
    single sheet or a list of sheets.

    Args:
        xlsx_path (str):
            The path to the Excel file to load.
        sheet_names (str | list | int | None, optional):
            Name or Index of the sheet in the Excel workbook to load.
            If 'None' then all sheets will be loaded.
            If 0 then first sheet in workbook will be loaded (this is the Default).
            If string then this is interpreted as the name of the sheet to load.
            If a list is passed, this can be a list of index values (int) or
            a list of strings with the sheet names to load.
            If multiple sheets are loaded they are concatenated into one data frame.
        usecols (list | str | None, optional):
            A list of columns to load, specified by general column names in Excel,
            e.g. usecols='B:D', usecols=['A', 'C', 'F']. Defaults to None (all columns).
        skip_rows (int | None, optional):
            Number of rows to skip on top of the sheet (e.g. to not read headlines).
            Defaults to None.
        header (int | None, optional):
            Excel Row (0-indexed) to use for the column labels of the parsed data frame.
            If file contains no header row, then you should explicitly pass header=None.
            Default is 0.
        names (list | None, optional):
            A list of column names to use. Default is None.
        na_values (list | None, optional):
            A list of values in the Excel that should become the Pandas NA value.
            Defaults to None.

    Returns:
        bool:
            False in case an error occurred, True otherwise.

    """

    if xlsx_path is not None and os.path.exists(xlsx_path):
        # Load data from Excel file
        try:
            df = pd.read_excel(
                io=xlsx_path,
                sheet_name=sheet_names,
                usecols=usecols,
                skiprows=skip_rows,
                header=header,
                names=names,
                na_values=na_values,
            )
            # If multiple sheets from an Excel workbook are loaded,
            # then read_excel() returns a dictionary. The keys are
            # the names of the sheets and the values are the data frames.
            # As this class can only handle one data frame per object,
            # We handle this case by concatenating the different sheets.
            # If you don't want this make sure your Excel workbook has only
            # one sheet or use the "sheet_name" parameter to select the one(s)
            # you want to load.
            if isinstance(df, dict):
                self.logger.info("Loading multiple Excel sheets from the workbook!")
                multi_sheet_df = pd.DataFrame()
                for sheet in df:
                    multi_sheet_df = pd.concat(
                        [multi_sheet_df, df[sheet]],
                        ignore_index=True,
                    )
                df = multi_sheet_df
            if self._df is None:
                self._df = df
            else:
                self._df = pd.concat([self._df, df], ignore_index=True)
        except FileNotFoundError:
            self.logger.error(
                "Excel file -> '%s' not found. Please check the file path.",
                xlsx_path,
            )
            return False
        except PermissionError:
            self.logger.error(
                "Missing permission to access the Excel file -> '%s'.",
                xlsx_path,
            )
            return False
        except OSError:
            self.logger.error(
                "An I/O error occurred while reading the Excel file -> '%s'",
                xlsx_path,
            )
            return False
        except ValueError:
            self.logger.error(
                "Invalid Excel input in file -> '%s'",
                xlsx_path,
            )
            return False
        except AttributeError:
            self.logger.error("Unexpected data structure in file -> %s", xlsx_path)
            return False
        except TypeError:
            self.logger.error("Unexpected data type in file -> %s", xlsx_path)
            return False
        except KeyError:
            self.logger.error("Missing key in Excel data in file -> %s", xlsx_path)
            return False

    else:
        self.logger.error(
            "Missing Excel file -> '%s'. You have not specified a valid path!",
            xlsx_path,
        )
        return False

    return True

save_excel_data

save_excel_data(
    excel_path: str,
    sheet_name: str = "Pandas Export",
    index: bool = False,
    columns: list | None = None,
) -> bool

Save the data frame to an Excel file, with robust error handling and logging.

Parameters:

Name	Type	Description	Default
excel_path	str	The file path to save the Excel file.	required
sheet_name	str	The sheet name where data will be saved. Default is 'Pandas Export'.	'Pandas Export'
index	bool	Whether to write the row names (index). Default is False.	False
columns	list | None	A list of column names to write into the Excel file. Defaults to None (all columns).	None

Returns:

Name	Type	Description
bool	bool	True = success, False = error.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def save_excel_data(
    self,
    excel_path: str,
    sheet_name: str = "Pandas Export",
    index: bool = False,
    columns: list | None = None,
) -> bool:
    """Save the data frame to an Excel file, with robust error handling and logging.

    Args:
        excel_path (str):
            The file path to save the Excel file.
        sheet_name (str, optional):
            The sheet name where data will be saved. Default is 'Pandas Export'.
        index (bool, optional):
            Whether to write the row names (index). Default is False.
        columns (list | None, optional):
            A list of column names to write into the Excel file. Defaults to None (all columns).

    Returns:
        bool:
            True = success, False = error.

    """

    try:
        # Check if the data frame is initialized first
        if self._df is None:
            self.logger.error(
                "Cannot write Excel file -> '%s' from empty / non-initialized data frame!", excel_path
            )
            return False

        # Check if the directory exists
        directory = os.path.dirname(excel_path)
        if directory and not os.path.exists(directory):
            os.makedirs(directory)

        # Validate columns if provided
        if columns:
            existing_columns = [col for col in columns if col in self._df.columns]
            missing_columns = set(columns) - set(existing_columns)
            if missing_columns:
                self.logger.warning(
                    "The following columns do not exist in the data frame and cannot be saved to Excel -> %s",
                    ", ".join(missing_columns),
                )
            columns = existing_columns

        # Attempt to save the data frame to Excel:
        self._df.to_excel(
            excel_path,
            sheet_name=sheet_name,
            index=index,
            columns=columns or None,  # Pass None if no columns provided
        )
        self.logger.info(
            "Data frame saved successfully to Excel file -> '%s'.",
            excel_path,
        )

    except FileNotFoundError as fnf_error:
        self.logger.error("Cannot write data frame to Excel file -> '%s'; error -> %s", excel_path, str(fnf_error))
        return False
    except PermissionError as pe:
        self.logger.error("Cannot write data frame to Excel file -> '%s'; error -> %s", excel_path, str(pe))
        return False
    except ValueError as ve:
        self.logger.error("Cannot write data frame to Excel file -> '%s'; error -> %s", excel_path, str(ve))
        return False
    except OSError as ose:
        self.logger.error("Cannot write data frame to Excel file -> '%s'; error -> %s", excel_path, str(ose))
        return False

    return True

load_csv_data

load_csv_data(
    csv_path: str,
    delimiter: str = ",",
    names: list | None = None,
    header: int | None = 0,
    usecols: list | None = None,
    encoding: str = "utf-8",
) -> bool

Load CSV (Comma separated values) data into data frame.

Parameters:

Name	Type	Description	Default
csv_path	str	The path to the CSV file or a URL (if the value starts with "http").	required
delimiter	str, optional, length = 1	The character used to delimit values. Default is "," (comma).	','
names	list | None	The list of column names. This is useful if file does not have a header line but just the data.	None
header	int | None	The index of the header line. Default is 0 (first line). None indicates that the file does not have a header line	0
usecols	list | None	There are three possible list values types: 1. int: These values are treated as column indices for columns to keep (first column has index 0). 2. str: The names of the columns to keep. For this to work the file needs either a header line (i.e. 'header != None') or the 'names' parameter must be specified. 3. bool: The length of the list must match the number of columns. Only columns that have a value of True are kept.	None
encoding	str	The encoding of the file. Default = "utf-8".	'utf-8'

Returns:

Name	Type	Description
bool	bool	False in case an error occurred, True otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_csv_data(
    self,
    csv_path: str,
    delimiter: str = ",",
    names: list | None = None,
    header: int | None = 0,
    usecols: list | None = None,
    encoding: str = "utf-8",
) -> bool:
    """Load CSV (Comma separated values) data into data frame.

    Args:
        csv_path (str):
            The path to the CSV file or a URL (if the value starts with "http").
        delimiter (str, optional, length = 1):
            The character used to delimit values. Default is "," (comma).
        names (list | None, optional):
            The list of column names. This is useful if file does not have a header line
            but just the data.
        header (int | None, optional):
            The index of the header line. Default is 0 (first line). None indicates
            that the file does not have a header line
        usecols (list | None, optional):
            There are three possible list values types:
            1. int:
                These values are treated as column indices for columns to keep
                (first column has index 0).
            2. str:
                The names of the columns to keep. For this to work the file needs
                either a header line (i.e. 'header != None') or the 'names'
                parameter must be specified.
            3. bool:
                The length of the list must match the number of columns. Only
                columns that have a value of True are kept.
        encoding (str, optional):
            The encoding of the file. Default = "utf-8".

    Returns:
        bool:
            False in case an error occurred, True otherwise.

    """

    if csv_path.startswith("http"):
        # Download file from remote location specified by the packageUrl
        # this must be a public place without authentication:
        self.logger.debug("Download CSV file from URL -> '%s'.", csv_path)

        try:
            response = requests.get(url=csv_path, timeout=1200)
            response.raise_for_status()
        except requests.exceptions.HTTPError:
            self.logger.error("HTTP error with -> %s", csv_path)
            return False
        except requests.exceptions.ConnectionError:
            self.logger.error("Connection error with -> %s", csv_path)
            return False
        except requests.exceptions.Timeout:
            self.logger.error("Timeout error with -> %s", csv_path)
            return False
        except requests.exceptions.RequestException:
            self.logger.error("Request error with -> %s", csv_path)
            return False

        self.logger.debug(
            "Successfully downloaded CSV file -> %s; status code -> %s",
            csv_path,
            response.status_code,
        )

        # Convert bytes to a string using utf-8 and create a file-like object
        csv_file = StringIO(response.content.decode(encoding))

    elif os.path.exists(csv_path):
        self.logger.debug("Using local CSV file -> '%s'.", csv_path)
        csv_file = csv_path

    else:
        self.logger.error(
            "Missing CSV file -> '%s' you have not specified a valid path!",
            csv_path,
        )
        return False

    # Load data from CSV file or buffer
    try:
        df = pd.read_csv(
            filepath_or_buffer=csv_file,
            delimiter=delimiter,
            names=names,
            header=header,
            usecols=usecols,
            encoding=encoding,
            skipinitialspace=True,
        )
        if self._df is None:
            self._df = df
        else:
            self._df = pd.concat([self._df, df], ignore_index=True)
    except FileNotFoundError:
        self.logger.error(
            "CSV file -> '%s' not found. Please check the file path.",
            csv_path,
        )
        return False
    except PermissionError:
        self.logger.error(
            "Permission denied to access the CSV file -> '%s'.",
            csv_path,
        )
        return False
    except OSError:
        self.logger.error("An I/O error occurred!")
        return False
    except ValueError:
        self.logger.error("Invalid CSV input in file -> %s", csv_path)
        return False
    except AttributeError:
        self.logger.error("Unexpected data structure in file -> %s", csv_path)
        return False
    except TypeError:
        self.logger.error("Unexpected data type in file -> %s", csv_path)
        return False
    except KeyError:
        self.logger.error("Missing key in CSV data -> %s", csv_path)
        return False

    return True

load_xml_data

load_xml_data(
    xml_path: str,
    xpath: str | None = None,
    xslt_path: str | None = None,
    encoding: str = "utf-8",
) -> bool

Load XML data into a Pandas data frame.

Parameters:

Name	Type	Description	Default
xml_path	str	The path to the XML file to load or a URL (if the value starts with "http").	required
xpath	str | None	An XPath to the elements we want to select. Defaults to None.	None
xslt_path	str | None	An XSLT transformation file to convert the XML data. Defaults to None.	None
encoding	str	The encoding of the file. Default is UTF-8.	'utf-8'

Returns:

Name	Type	Description
bool	bool	False in case an error occurred, True otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_xml_data(
    self,
    xml_path: str,
    xpath: str | None = None,
    xslt_path: str | None = None,
    encoding: str = "utf-8",
) -> bool:
    """Load XML data into a Pandas data frame.

    Args:
        xml_path (str):
            The path to the XML file to load or a URL (if the value starts with "http").
        xpath (str | None, optional):
            An XPath to the elements we want to select. Defaults to None.
        xslt_path (str | None, optional):
            An XSLT transformation file to convert the XML data. Defaults to None.
        encoding (str, optional):
            The encoding of the file. Default is UTF-8.

    Returns:
        bool:
            False in case an error occurred, True otherwise.

    """

    if xml_path.startswith("http"):
        # Download file from remote location specified by the packageUrl
        # this must be a public place without authentication:
        self.logger.debug("Download XML file from URL -> '%s'.", xml_path)

        try:
            response = requests.get(url=xml_path, timeout=1200)
            response.raise_for_status()
        except requests.exceptions.HTTPError:
            self.logger.error("HTTP error with -> %s", xml_path)
            return False
        except requests.exceptions.ConnectionError:
            self.logger.error("Connection error with -> %s", xml_path)
            return False
        except requests.exceptions.Timeout:
            self.logger.error("Timeout error with -> %s", xml_path)
            return False
        except requests.exceptions.RequestException:
            self.logger.error("Request error with -> %s", xml_path)
            return False

        self.logger.debug(
            "Successfully downloaded XML file -> '%s'; status code -> %s",
            xml_path,
            response.status_code,
        )
        # Convert bytes to a string using utf-8 and create a file-like object
        xml_file = StringIO(response.content.decode(encoding))

    elif os.path.exists(xml_path):
        self.logger.debug("Using local XML file -> '%s'.", xml_path)
        xml_file = xml_path

    else:
        self.logger.error(
            "Missing XML file -> '%s'. You have not specified a valid path or URL!",
            xml_path,
        )
        return False

    # Load data from XML file or buffer
    try:
        df = pd.read_xml(
            path_or_buffer=xml_file,
            xpath=xpath,
            stylesheet=xslt_path,
            encoding=encoding,
        )
        # Process the loaded data as needed
        if self._df is None:
            self._df = df
        else:
            self._df = pd.concat([self._df, df], ignore_index=True)
        self.logger.info("XML file -> '%s' loaded successfully!", xml_path)
    except FileNotFoundError:
        self.logger.error("XML file -> '%s' not found.", xml_path)
        return False
    except PermissionError:
        self.logger.error(
            "Missing permission to access the XML file -> '%s'.",
            xml_path,
        )
        return False
    except OSError:
        self.logger.error("An I/O error occurred loading from -> %s", xml_path)
        return False
    except ValueError:
        self.logger.error("Invalid XML data in file -> %s", xml_path)
        return False
    except AttributeError:
        self.logger.error("Unexpected data structure in XML file -> %s", xml_path)
        return False
    except TypeError:
        self.logger.error("Unexpected data type in XML file -> %s", xml_path)
        return False
    except KeyError:
        self.logger.error("Missing key in XML file -> %s", xml_path)
        return False

    return True

load_directory

load_directory(path_to_root: str) -> bool

Load directory structure into Pandas data frame.

Parameters:

Name	Type	Description	Default
path_to_root	str	Path to the root element of the directory structure.	required

Returns:

Name	Type	Description
bool	bool	True = Success, False = Failure

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_directory(self, path_to_root: str) -> bool:
    """Load directory structure into Pandas data frame.

    Args:
        path_to_root (str):
            Path to the root element of the directory structure.

    Returns:
        bool: True = Success, False = Failure

    """

    try:
        # Check if the provided path is a directory
        if not os.path.isdir(path_to_root):
            self.logger.error(
                "The provided path -> '%s' is not a valid directory.",
                path_to_root,
            )
            return False

        # Initialize a list to hold file information
        data = []

        # Walk through the directory
        for root, _, files in os.walk(path_to_root):
            for file in files:
                file_path = os.path.join(root, file)
                file_size = os.path.getsize(file_path)
                relative_path = os.path.relpath(file_path, path_to_root)
                path_parts = relative_path.split(os.sep)

                # Create a dictionary with the path parts and file details
                entry = {"level {}".format(i): part for i, part in enumerate(path_parts[:-1], start=1)}

                entry.update(
                    {
                        "filename": path_parts[-1],
                        "size": file_size,
                        "path": path_parts[1:-1],
                        "relative_path": relative_path,
                        "download_dir": root,
                    },
                )
                data.append(entry)

        # Create data frame from list of dictionaries:
        self._df = pd.DataFrame(data)

        # Determine the maximum number of levels
        max_levels = max((len(entry) - 2 for entry in data), default=0)

        # Ensure all entries have the same number of levels:
        for entry in data:
            for i in range(1, max_levels + 1):
                entry.setdefault("level {}".format(i), "")

        # Convert to data frame again to make sure all columns are consistent:
        self._df = pd.DataFrame(data)

    except NotADirectoryError:
        self.logger.error(
            "Provided path -> '%s' is not a directory!",
            path_to_root,
        )
        return False
    except FileNotFoundError:
        self.logger.error(
            "Provided path -> '%s' does not exist in file system!",
            path_to_root,
        )
        return False
    except PermissionError:
        self.logger.error(
            "Permission error accessing path -> '%s'!",
            path_to_root,
        )
        return False

    return True

load_xml_directory

load_xml_directory(
    path_to_root: str,
    xpath: str | None = None,
    xml_files: list | None = None,
) -> bool

Load XML files from a directory structure into Pandas data frame.

Parameters:

Name	Type	Description	Default
path_to_root	str	Path to the root element of the directory structure.	required
xpath	str | None	XPath to the XML elements we want to select. Defaults to None.	None
xml_files	list | None	Names of the XML files to load from the directory. If None or empty then ["docovw.xml"] is used.	None

Returns:

Name	Type	Description
bool	bool	True = Success, False = Failure

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_xml_directory(
    self,
    path_to_root: str,
    xpath: str | None = None,
    xml_files: list | None = None,
) -> bool:
    """Load XML files from a directory structure into Pandas data frame.

    Args:
        path_to_root (str):
            Path to the root element of the directory structure.
        xpath (str | None, optional):
            XPath to the XML elements we want to select. Defaults to None.
        xml_files (list | None, optional):
            Names of the XML files to load from the directory.
            If None or empty then ["docovw.xml"] is used.

    Returns:
        bool:
            True = Success, False = Failure

    """

    # Establish a default if None is passed via the parameter:
    if not xml_files:
        xml_files = ["docovw.xml"]

    try:
        # Check if the provided path is a directory
        if not os.path.isdir(path_to_root):
            self.logger.error(
                "The provided path -> '%s' is not a valid directory.",
                path_to_root,
            )
            return False

        # Walk through the directory
        for root, _, files in os.walk(path_to_root):
            for file in files:
                file_path = os.path.join(root, file)
                file_size = os.path.getsize(file_path)
                file_name = os.path.basename(file_path)

                if file_name in xml_files:
                    self.logger.info(
                        "Load XML file -> '%s' of size -> %s from -> '%s'...",
                        file_name,
                        file_size,
                        file_path,
                    )
                    success = self.load_xml_data(file_path, xpath=xpath)
                    if success:
                        self.logger.info(
                            "Successfully loaded XML file -> '%s'.",
                            file_path,
                        )

    except NotADirectoryError:
        self.logger.error(
            "Provided path -> '%s' is not a directory",
            path_to_root,
        )
        return False
    except FileNotFoundError:
        self.logger.error(
            "Provided path -> '%s' does not exist in file system!",
            path_to_root,
        )
        return False
    except PermissionError:
        self.logger.error(
            "Missing permission to access path -> '%s'",
            path_to_root,
        )
        return False

    return True

load_web_links

load_web_links(
    url: str,
    common_data: dict | None = None,
    pattern: str = "",
) -> list | None

Get all linked file URLs on a given web page (url) that are following a given pattern.

Construct a list of dictionaries based on this. This method is a helper method for load_web() below.

Parameters:

Name	Type	Description	Default
url	str	The web page URL.	required
common_data	dict | None	Fields that should be added to each dictionary item. Defaults to None.	None
pattern	str	Regular Expression. Defaults to r"".	''

Returns:

Type	Description
list | None	list | None: List of dictionaries (one per link on the web page that complies with the given regular expression). The list is empty if the page cannot be retrieved or does not contain any links.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_web_links(
    self,
    url: str,
    common_data: dict | None = None,
    pattern: str = r"",
) -> list | None:
    """Get all linked file URLs on a given web page (url) that are following a given pattern.

    Construct a list of dictionaries based on this. This method is a helper method for load_web() below.

    Args:
        url (str):
            The web page URL.
        common_data (dict | None, optional):
            Fields that should be added to each dictionary item. Defaults to None.
        pattern (str, optional):
            Regular Expression. Defaults to r"".

    Returns:
        list | None:
            List of dictionaries (one per link on the web page that complies with the given
            regular expression). The list is empty if the page cannot be retrieved
            or does not contain any links.

    """

    try:
        response = requests.get(url, timeout=300)
        response.raise_for_status()
    except requests.RequestException:
        self.logger.error("Failed to retrieve page at %s", url)
        return []

    # Find all file links (hyperlinks) on the page (no file extension assumed)
    # Example filename pattern: "al022023.public.005"
    file_links = re.findall(r'href="([^"]+)"', response.text)
    if not file_links:
        self.logger.warning("No file links found on the web page -> %s", url)
        return []

    result_list = []
    base_url = url if url.endswith("/") else url + "/"

    for link in file_links:
        data = common_data.copy() if common_data else {}

        # Construct the full URL
        full_url = base_url + link.lstrip("/")

        if pattern:
            # Filter by expected naming pattern for links
            match = re.search(pattern, link)
            if not match:
                continue

            # Extract and assign groups if they exist
            # TODO(mdiefenb): these names are currently hard-coded
            # for the National Hurricane Center Dataset (NHC)
            if len(match.groups()) >= 1:
                data["Code"] = match.group(1).upper()
            if len(match.groups()) >= 2:
                data["Type"] = match.group(2)
            if len(match.groups()) >= 3:
                data["Message ID"] = match.group(3)

        data["URL"] = full_url
        data["Filename"] = link

        result_list.append(data)

    return result_list

load_web

load_web(
    values: list,
    value_name: str,
    url_templates: list,
    special_values: list | None = None,
    special_url_templates: dict | None = None,
    pattern: str = "",
) -> bool

Traverse years and bulletin types to collect all bulletin URLs.

Parameters:

Name	Type	Description	Default
values	list	List of values to traverse over.	required
value_name	str	Dictionary key to construct an item in combination with a value from values	required
url_templates	list	URLs to traverse per value. The URLs should contain one {} that is replaced by the current value.	required
special_values	list | None	List of values (a subset of the other values list) that we want to handle in a special way. Defaults to None.	None
special_url_templates	dict | None	URLs for the special values. Defaults to None. The dictionary keys are the special values. The dictionary values are lists of special URLs with placeholders.	None
pattern	str	Regular expression to find the proper links on the page. Defaults to r"".	''

Returns:

Name	Type	Description
bool	bool	True for success, False in case of an error.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def load_web(
    self,
    values: list,
    value_name: str,
    url_templates: list,
    special_values: list | None = None,
    special_url_templates: dict | None = None,
    pattern: str = r"",
) -> bool:
    """Traverse years and bulletin types to collect all bulletin URLs.

    Args:
        values (list):
            List of values to traverse over.
        value_name (str):
            Dictionary key to construct an item in combination with a value from values
        url_templates (list):
            URLs to traverse per value. The URLs should contain one {} that is
            replaced by the current value.
        special_values (list | None, optional):
            List of values (a subset of the other values list)
            that we want to handle in a special way. Defaults to None.
        special_url_templates (dict | None, optional):
            URLs for the special values. Defaults to None.
            The dictionary keys are the special values. The
            dictionary values are lists of special URLs with placeholders.
        pattern (str, optional):
            Regular expression to find the proper links on the page. Defaults to r"".

    Returns:
        bool:
            True for success, False in case of an error.

    """

    result_list = []

    # We have two nested for loops below. The outer one traverses over all placeholder values.
    # These could be the calendar years, e.g. [2003,...,2024]
    # The inner for loop traverses over the list of specified URLs. We can have multiple for
    # each value.

    # Do we have a list of placeholder values we want to iterate over?
    if values:
        # Traverse all values in the values list:
        for value in values:
            # Do we want a special treatment for this value (e.g. the current year)
            if special_values and value in special_values:
                self.logger.debug("Processing special value -> '%s'...", value)
                if value not in special_url_templates and str(value) not in special_url_templates:
                    self.logger.error(
                        "Cannot find key -> '%s' in special URL templates dictionary -> %s! Skipping...",
                        value,
                        str(special_url_templates),
                    )
                    continue
                # If the dictionary uses string keys then we need to convert the value
                # to a string as well to avoid key errors:
                if str(value) in special_url_templates:
                    value = str(value)
                special_url_template_list = special_url_templates[value]
                for special_url_template in special_url_template_list:
                    # Now the value is inserted into the placeholder in the URL:
                    special_url = special_url_template.format(value)
                    common_data = {value_name: value} if value_name else None
                    result_list += self.load_web_links(
                        url=special_url,
                        common_data=common_data,
                        pattern=pattern,
                    )
            else:  # normal URLs
                self.logger.info("Processing value -> '%s'...", value)
                for url_template in url_templates:
                    # Now the value is inserted into the placeholder in the URL:
                    url = url_template.format(value)
                    common_data = {value_name: value} if value_name else None
                    result_list += self.load_web_links(
                        url=url,
                        common_data=common_data,
                        pattern=pattern,
                    )
    elif not values:
        for url_template in url_templates:
            result_list += self.load_web_links(
                url=url_template,
                common_data=None,
                pattern=pattern,
            )

    # Add the data list to the data frame:
    self.append(result_list)

    return True

partitionate

partitionate(number: int) -> list

Partition a data frame into equally sized partitions.

Parameters:

Name	Type	Description	Default
number	int	The number of desired partitions.	required

Returns:

Name	Type	Description
list	list	A list of created partitions.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def partitionate(self, number: int) -> list:
    """Partition a data frame into equally sized partitions.

    Args:
        number (int):
            The number of desired partitions.

    Returns:
        list:
            A list of created partitions.

    """

    # Calculate the approximate size of each partition
    size = len(self._df)

    if size >= number:
        partition_size = size // number
        remainder = size % number
    else:
        partition_size = size
        number = 1
        remainder = 0

    self.logger.info(
        "Data frame has -> %s elements. We split it into -> %s partitions with -> %s row%s and remainder -> %s...",
        str(size),
        str(number),
        str(partition_size),
        "s" if partition_size > 1 else "",
        str(remainder),
    )

    # Initialize a list to store partitions:
    partitions = []
    start_index = 0

    # Slice the data frame into equally sized partitions:
    for i in range(number):
        # Calculate the end index for this partition
        end_index = start_index + partition_size + (1 if i < remainder else 0)
        partition = self._df.iloc[start_index:end_index]
        partitions.append(partition)
        start_index = end_index

    return partitions

partitionate_by_column

partitionate_by_column(column_name: str) -> list | None

Partition a data frame based on equal values in a specified column.

Parameters:

Name	Type	Description	Default
column_name	str	The column name to partition by.	required

Returns:

Type	Description
list | None	list | None: List of partitions or None in case of an error (e.g. column name does not exist).

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def partitionate_by_column(self, column_name: str) -> list | None:
    """Partition a data frame based on equal values in a specified column.

    Args:
        column_name (str):
            The column name to partition by.

    Returns:
        list | None:
            List of partitions or None in case of an error (e.g. column name does not exist).

    """

    if column_name not in self._df.columns:
        self.logger.error(
            "Cannot partitionate by column -> '%s'. Column does not exist in the data frame. Data frame has these columns -> %s",
            column_name,
            str(self._df.columns),
        )
        return None

    # Separate rows with NaN or None values in the specified column:
    nan_partitions = self._df[self._df[column_name].isna()]

    # Keep only rows where the specified column has valid (non-NaN) values:
    non_nan_df = self._df.dropna(subset=[column_name])

    # Group the non-NaN DataFrame by the specified column's values:
    grouped = non_nan_df.groupby(column_name)

    # Create a list of partitions (DataFrames) for each unique value in the column:
    partitions = [group for _, group in grouped]

    # Add each row with NaN/None as its own partition
    # iterrows() returns each row as a Series. To convert it back to a DataFrame:
    # 1. .to_frame() turns the Series into a DataFrame, but with the original column names as rows.
    # 2. .T (transpose) flips it back, turning the original row into a proper DataFrame row.
    # This ensures that even rows with NaN values are treated as DataFrame partitions.
    partitions.extend([row.to_frame().T for _, row in nan_partitions.iterrows()])

    self.logger.info(
        "Data frame has been partitioned into -> %s partitions based on the values in column -> '%s'...",
        str(len(partitions)),
        column_name,
    )

    return partitions

deduplicate

deduplicate(
    unique_fields: list, inplace: bool = True
) -> DataFrame

Remove duplicate rows that have all fields in unique_fields in common.

The index defined for this Data object (index_columns) is kept - its rows are selected by it (e.g. with loc[]). Without one, the rows are numbered again (0, 1, 2, ...).

Parameters:

Name	Type	Description	Default
unique_fields	list	Defines the fields for which we want a unique combination for.	required
inplace	bool	True if the deduplication happens in-place. Defaults to True.	True

Returns:

Type	Description
DataFrame	pd.DataFrame: If inplace is False then a new deduplicated data frame is returned. Otherwise the object is modified in place and self._df is returned.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def deduplicate(self, unique_fields: list, inplace: bool = True) -> pd.DataFrame:
    """Remove duplicate rows that have all fields in unique_fields in common.

    The index defined for this Data object (`index_columns`) is kept - its rows are selected by it (e.g.
    with `loc[]`). Without one, the rows are numbered again (0, 1, 2, ...).

    Args:
        unique_fields (list):
            Defines the fields for which we want a unique combination for.
        inplace (bool, optional):
            True if the deduplication happens in-place. Defaults to True.

    Returns:
        pd.DataFrame:
            If inplace is False then a new deduplicated data frame is returned.
            Otherwise the object is modified in place and self._df is returned.

    """

    if inplace:
        self._df.drop_duplicates(subset=unique_fields, inplace=True)
        if not self._index_columns:
            self._df.reset_index(drop=True, inplace=True)
        return self._df
    else:
        df = self._df.drop_duplicates(subset=unique_fields, inplace=False)
        if not self._index_columns:
            df = df.reset_index(drop=True, inplace=False)
        return df

sort

sort(
    sort_fields: list,
    inplace: bool = True,
    ascending: bool = True,
) -> DataFrame | None

Sort the data frame based on one or multiple fields.

Sorting can be either in place or return it as a new data frame (e.g. not modifying self._df).

Parameters:

Name	Type	Description	Default
sort_fields	list	The columns / fields to be used for sorting.	required
inplace	bool	If the sorting should be inplace, i.e. modifying self._df. Defaults to True.	True
ascending	bool	If True, sort in ascending order. If False, sort in descending order. Defaults to True.	True

Returns:

Type	Description
DataFrame | None	pd.DataFrame | None: New data frame (if inplace = False) or self._df (if inplace = True). None in case of an error.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def sort(self, sort_fields: list, inplace: bool = True, ascending: bool = True) -> pd.DataFrame | None:
    """Sort the data frame based on one or multiple fields.

    Sorting can be either in place or return it as a new data frame
    (e.g. not modifying self._df).

    Args:
        sort_fields (list):
            The columns / fields to be used for sorting.
        inplace (bool, optional):
            If the sorting should be inplace, i.e. modifying self._df.
            Defaults to True.
        ascending (bool, optional):
            If True, sort in ascending order. If False, sort in descending order. Defaults to True.

    Returns:
        pd.DataFrame | None:
            New data frame (if inplace = False) or self._df (if inplace = True).
            None in case of an error.

    """

    if self._df is None:
        return None

    if not all(sort_field in self._df.columns for sort_field in sort_fields):
        self.logger.warning(
            "Not all of the given sort fields -> %s do exist in the data frame.",
            str(sort_fields),
        )
        # Reduce the sort fields to those that really exist in the data frame:
        sort_fields = [sort_field for sort_field in sort_fields if sort_field in self._df.columns]
        self.logger.warning(
            "Only these given sort fields -> %s do exist as columns in the data frame.",
            str(sort_fields),
        )

    if inplace:
        self._df.sort_values(by=sort_fields, inplace=True, ascending=ascending)
        self._df.reset_index(drop=True, inplace=True)
        return self._df
    else:
        df = self._df.sort_values(by=sort_fields, inplace=False, ascending=ascending)
        df = df.reset_index(drop=True, inplace=False)
        return df

flatten

flatten(
    parent_field: str,
    flatten_fields: list,
    concatenator: str = "_",
) -> None

Flatten a sub-dictionary by copying selected fields to the parent dictionary.

This is e.g. useful to then de-duplicate a data frame. To flatten a data frame makes sense in situation when a column used to have a list of dictionaries and got "exploded" (see explode_and_flatten() method below). In this case the column has dictionary values that then can be flattened.

Parameters:

Name	Type	Description	Default
parent_field	str	Name of the column that has the dictionary values. It is also used as name prefix of the new columns in the data frame. The flattened field names are appended using the concatenator.	required
flatten_fields	list	Fields in the dictionary of the source column that are copied as new columns into the data frame.	required
concatenator	str	Character or string used to concatenate the parent field with the flattened field to create a unique name. Defaults to "_".	'_'

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def flatten(self, parent_field: str, flatten_fields: list, concatenator: str = "_") -> None:
    """Flatten a sub-dictionary by copying selected fields to the parent dictionary.

    This is e.g. useful to then de-duplicate a data frame.
    To flatten a data frame makes sense in situation when a column used
    to have a list of dictionaries and got "exploded" (see explode_and_flatten()
    method below). In this case the column has dictionary values that then can
    be flattened.

    Args:
        parent_field (str):
            Name of the column that has the dictionary values. It is also used as name
            prefix of the new columns in the data frame. The flattened field
            names are appended using the concatenator.
        flatten_fields (list):
            Fields in the dictionary of the source column that are copied
            as new columns into the data frame.
        concatenator (str, optional):
            Character or string used to concatenate the parent field with the flattened field
            to create a unique name. Defaults to "_".

    """

    # First do a sanity check if the data frame is not yet initialized.
    if self._df is None:
        self.logger.error(
            "The data frame is not initialized or empty. Cannot flatten field(s) -> '%s' in the data frame.",
            flatten_fields,
        )
        return

    if parent_field not in self._df.columns:
        self.logger.warning(
            "The parent field -> '%s' cannot be flattened as it doesn't exist as column in the data frame!",
            parent_field,
        )
        return

    for flatten_field in flatten_fields:
        flat_field = parent_field + concatenator + flatten_field
        # The following expression generates a new column in the
        # data frame with the name of 'flat_field'.
        # In the lambda function x is a dictionary that includes the subvalues
        # and it returns the value of the given flatten field
        # (if it exists, otherwise None). So x is self._df[parent_field], i.e.
        # what the lambda function gets 'applied' on.
        self._df[flat_field] = self._df[parent_field].apply(
            lambda x, sub_field=flatten_field: x.get(sub_field, None) if isinstance(x, dict) else None,
        )

explode_and_flatten

explode_and_flatten(
    explode_fields: str | list,
    flatten_fields: list | None = None,
    make_unique: bool = False,
    reset_index: bool = False,
    split_string_to_list: bool = False,
    separator: str = ";,",
) -> DataFrame | None

Explode a substructure in the Pandas data frame.

Parameters:

Name	Type	Description	Default
explode_fields	str | list	Field(s) to explode. Each field to explode should have a list structure. Exploding multiple columns at once is possible. This delivers a very different result compared to exploding one column after the other!	required
flatten_fields	list | None	Fields in the exploded substructure to include in the main dictionaries for easier processing. Defaults to None.	None
make_unique	bool	If True, deduplicate the exploded data frame (only done if flatten_fields are given). Defaults to False.	False
reset_index	bool	If True, then the index is reset, False = Index is not reset. Defaults to False.	False
split_string_to_list	bool	If True, the string values in the columns to explode are split into lists (using the separator) before the columns are exploded. Defaults to False.	False
separator	str	Characters used to split the string values in the given column into a list. Each character of the string is treated as a separate separator. Defaults to ";,".	';,'

Returns:

Type	Description
DataFrame | None	pd.DataFrame | None: Pointer to the Pandas data frame.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def explode_and_flatten(
    self,
    explode_fields: str | list,
    flatten_fields: list | None = None,
    make_unique: bool = False,
    reset_index: bool = False,
    split_string_to_list: bool = False,
    separator: str = ";,",
) -> pd.DataFrame | None:
    """Explode a substructure in the Pandas data frame.

    Args:
        explode_fields (str | list):
            Field(s) to explode. Each field to explode should have a list structure.
            Exploding multiple columns at once is possible. This delivers
            a very different result compared to exploding one column after the other!
        flatten_fields (list | None, optional):
            Fields in the exploded substructure to include
            in the main dictionaries for easier processing. Defaults to None.
        make_unique (bool, optional):
            If True, deduplicate the exploded data frame (only done if flatten_fields
            are given). Defaults to False.
        reset_index (bool, optional):
            If True, then the index is reset, False = Index is not reset. Defaults to False.
        split_string_to_list (bool, optional):
            If True, the string values in the columns to explode are split into lists
            (using the separator) before the columns are exploded. Defaults to False.
        separator (str, optional):
            Characters used to split the string values in the given column into a list.
            Each character of the string is treated as a separate separator. Defaults to ";,".

    Returns:
        pd.DataFrame | None:
            Pointer to the Pandas data frame.

    """

    def update_column(row: pd.Series, sub: str) -> str:
        """Extract the value of a sub-column from a nested dictionary within a Pandas Series.

        Args:
            row (pd.Series):
                A row from the data frame.
            sub (str):
                The sub-column name to extract.

        Returns:
            str:
                The value of the sub-column, or an empty string if not found.

        """

        if isinstance(row, dict) and sub in row:
            return row[sub]
        return ""

    # end def update_column()

    def string_to_list(value: str) -> list:
        """Convert a string to a list by splitting it using a specified separator.

        If the input is already a list, it is returned as-is. If the input is `None` or a missing value,
        an empty list is returned. Otherwise, the string is split into a list of substrings using
        the given separator. Leading and trailing spaces in the resulting substrings are removed.

        Args:
            value (str):
                The input string to be converted into a list. Can also be a list, `None`,
                or a missing value (e.g., NaN).

        Returns:
            list:
                A list of substrings if the input is a string, or an empty list if the input
                is `None` or a missing value. If the input is already a list, it is returned unchanged.

        """

        # Check if the value is already a list; if so, return it directly
        if isinstance(value, list):
            return value

        # If the value is None or a missing value (e.g., NaN), return an empty list
        if not value or pd.isna(value):
            return []

        # Use a regular expression to split the string by the separator
        # and remove leading/trailing spaces from each resulting substring
        return_list = re.split(rf"[{separator}]\s*", str(value))

        return return_list

    # end def string_to_list()

    #
    # Start of main method:
    #

    # First do a sanity check if the data frame is not yet initialized.
    if self._df is None:
        self.logger.error(
            "The data frame is not initialized or empty. Cannot explode data frame.",
        )
        return None

    # Next do a sanity check for the given explode_field. It should
    # either be a string (single column name) or a list (multiple column names):
    if isinstance(explode_fields, list):
        self.logger.info("Exploding list of columns -> %s", str(explode_fields))
    elif isinstance(explode_fields, str):
        self.logger.info("Exploding single column -> '%s'", explode_fields)
    else:
        self.logger.error(
            "Illegal explode field(s) data type -> %s. Explode field must either be a string or a list of strings.",
            type(explode_fields),
        )
        return self._df

    # Ensure explode_fields is a list for uniform processing:
    if isinstance(explode_fields, str):
        explode_fields = [explode_fields]

    # Process nested field names with '.'
    processed_fields = []
    for field in explode_fields:
        # The "." indicates that the column has dictionary values:
        if "." in field:
            main, sub = field.split(".", 1)
            if main not in self._df.columns:
                self.logger.error(
                    "The column -> '%s' does not exist in the data frame! Cannot explode it. Data frame has these columns -> %s",
                    main,
                    str(self._df.columns.tolist()),
                )
                continue

            # Use update_column to extract the dictionary key specified by the sub value:
            self.logger.info(
                "Extracting dictionary value for key -> '%s' from column -> '%s'.",
                sub,
                main,
            )
            self._df[main] = self._df[main].apply(update_column, args=(sub,))
            processed_fields.append(main)
        else:
            processed_fields.append(field)

    # Verify all processed fields exist in the data frame:
    missing_columns = [col for col in processed_fields if col not in self._df.columns]
    if missing_columns:
        self.logger.error(
            "The following columns are missing in the data frame and cannot be exploded -> %s. Data frame has these columns -> %s",
            missing_columns,
            str(self._df.columns.tolist()),
        )
        return self._df

    # Handle splitting strings into lists if required:
    if split_string_to_list:
        for field in processed_fields:
            self.logger.info(
                "Splitting strings in column -> '%s' into lists using separator -> '%s'",
                field,
                separator,
            )
            # Apply the function to convert the string values in the column (give by the name in explode_field) to lists
            # The string_to_list() sub-method above also considers the separator parameter.
            self._df[field] = self._df[field].apply(string_to_list)

    # Explode all specified columns at once.
    # explode() can either take a string field or a list of fields.
    # # It is VERY important to do the explosion of multiple columns together -
    # otherwise we get combinatorial explosion. Explosion of multiple columns 1-by-1
    # is VERY different from doing the explosion together!
    self.logger.info("Validated column(s) to explode -> %s", processed_fields)
    try:
        self._df = self._df.explode(
            column=processed_fields,
            ignore_index=reset_index,
        )
    except ValueError:
        self.logger.error(
            "Error exploding columns -> %s",
            processed_fields,
        )
        return self._df

    if flatten_fields:
        # Ensure that flatten() is called for each exploded column
        for field in processed_fields:
            self.flatten(parent_field=field, flatten_fields=flatten_fields)

        # Deduplicate rows if required
        if make_unique:
            self._df.drop_duplicates(subset=flatten_fields, inplace=True)

    # Reset index explicitly if not handled during explode
    if reset_index:
        self._df.reset_index(drop=True, inplace=True)

    return self._df

drop_columns

drop_columns(
    column_names: list, inplace: bool = True
) -> DataFrame

Drop selected columns from the Pandas data frame.

Parameters:

Name	Type	Description	Default
column_names	list	The list of column names to drop.	required
inplace	bool	Whether or not the dropping should be inplace, i.e. modifying self._df. Defaults to True.	True

Returns:

Type	Description
DataFrame	pd.DataFrame: New data frame (if inplace = False) or self._df (if inplace = True)

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def drop_columns(self, column_names: list, inplace: bool = True) -> pd.DataFrame:
    """Drop selected columns from the Pandas data frame.

    Args:
        column_names (list):
            The list of column names to drop.
        inplace (bool, optional):
            Whether or not the dropping should be inplace, i.e. modifying self._df.
            Defaults to True.

    Returns:
        pd.DataFrame:
            New data frame (if inplace = False) or self._df (if inplace = True)

    """

    if not all(column_name in self._df.columns for column_name in column_names):
        # Reduce the column names to those that really exist in the data frame:
        column_names = [column_name for column_name in column_names if column_name in self._df.columns]
        self.logger.info(
            "Drop columns -> %s from the data frame.",
            str(column_names),
        )

    if inplace:
        self._df.drop(column_names, axis=1, inplace=True)
        return self._df
    else:
        df = self._df.drop(column_names, axis=1, inplace=False)
        return df

keep_columns

keep_columns(
    column_names: list, inplace: bool = True
) -> DataFrame

Keep only selected columns in the data frame. Drop the rest.

Parameters:

Name	Type	Description	Default
column_names	list	A list of column names to keep.	required
inplace	bool	If the keeping should be inplace, i.e. modifying self._df. Defaults to True.	True

Returns:

Type	Description
DataFrame	pd.DataFrame: New data frame (if inplace = False) or self._df (if inplace = True). None if inplace = False and none of the given column names exist in the data frame.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def keep_columns(self, column_names: list, inplace: bool = True) -> pd.DataFrame:
    """Keep only selected columns in the data frame. Drop the rest.

    Args:
        column_names (list):
            A list of column names to keep.
        inplace (bool, optional):
            If the keeping should be inplace, i.e. modifying self._df.
            Defaults to True.

    Returns:
        pd.DataFrame:
            New data frame (if inplace = False) or self._df (if inplace = True).
            None if inplace = False and none of the given column names exist in the data frame.

    """

    if not all(column_name in self._df.columns for column_name in column_names):
        # Reduce the column names to those that really exist in the data frame:
        column_names = [column_name for column_name in column_names if column_name in self._df.columns]
        self.logger.info(
            "Reduce columns to keep to these columns -> %s that do exist in the data frame.",
            column_names,
        )

    if inplace:
        # keep only those columns which are in column_names:
        if column_names != []:
            self._df = self._df[column_names]
        return self._df
    else:
        # keep only those columns which are in column_names:
        if column_names != []:
            df = self._df[column_names]
            return df
        return None

rename_column

rename_column(
    old_column_name: str, new_column_name: str
) -> bool

Rename a data frame column.

Parameters:

Name	Type	Description	Default
old_column_name	str	The old name of the column.	required
new_column_name	str	The new name of the column.	required

Returns:

Name	Type	Description
bool	bool	True = Success, False = Error

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def rename_column(self, old_column_name: str, new_column_name: str) -> bool:
    """Rename a data frame column.

    Args:
        old_column_name (str):
            The old name of the column.
        new_column_name (str):
            The new name of the column.

    Returns:
        bool:
            True = Success, False = Error

    """

    if self._df is None:
        return False

    if old_column_name not in self._df.columns:
        self.logger.error(
            "Cannot rename column -> '%s'. It does not exist in the data frame! Data frame has these columns -> %s",
            old_column_name,
            str(self._df.columns),
        )
        return False

    if new_column_name in self._df.columns:
        self.logger.error(
            "Cannot rename column -> '%s' to -> '%s'. New name does already exist as column in the data frame! Data frame has these columns -> %s",
            old_column_name,
            new_column_name,
            str(self._df.columns),
        )
        return False

    self._df.rename(columns={old_column_name: new_column_name}, inplace=True)

    return True

is_dict_column

is_dict_column(
    column: Series, threshold: float = 0.5
) -> bool

Safely checks if a column predominantly contains dictionary-like objects.

Parameters:

Name	Type	Description	Default
column	Series	The pandas Series (column) to check. If not a Series, returns False and logs an error.	required
threshold	float	0.0 < threshold <= 1.0. Float representation of the percentage. Default = 0.5 (50%). If out of bounds, returns False and logs an error.	0.5

Returns:

Name	Type	Description
bool	bool	True if the column contains mostly dictionary-like objects, False otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def is_dict_column(self, column: pd.Series, threshold: float = 0.5) -> bool:
    """Safely checks if a column predominantly contains dictionary-like objects.

    Args:
        column (pd.Series):
            The pandas Series (column) to check. If not a Series, returns False and logs an error.
        threshold (float, optional):
            0.0 < threshold <= 1.0. Float representation of the percentage.
            Default = 0.5 (50%). If out of bounds, returns False and logs an error.

    Returns:
        bool:
            True if the column contains mostly dictionary-like objects, False otherwise.

    """

    if not isinstance(column, pd.Series):
        self.logger.error(
            "Expected Pandas series, but got -> %s",
            str(type(column)),
        )
        return False
    if not 0.0 < threshold <= 1.0:
        self.logger.error(
            "Threshold must be between 0.0 and 1.0, but got -> %s",
            str(threshold),
        )
        return False

    # Drop null values (NaN or None) and check types of remaining values
    non_null_values = column.dropna()
    dict_count = non_null_values.apply(lambda x: isinstance(x, dict)).sum()

    # If more than threshold % of non-null values are dictionaries, return True.
    # Else return False.
    return dict_count / len(non_null_values) > threshold if len(non_null_values) > 0 else False

is_list_column

is_list_column(
    column: Series, threshold: float = 0.5
) -> bool

Safely checks if a column predominantly contains list-like objects.

Parameters:

Name	Type	Description	Default
column	Series	The pandas Series (column) to check. If not a Series, returns False and logs an error.	required
threshold	float	0.0 < threshold <= 1.0. Float representation of the percentage. Default = 0.5 (50%). If out of bounds, returns False and logs an error.	0.5

Returns:

Name	Type	Description
bool	bool	True if the column contains mostly list-like objects, False otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def is_list_column(self, column: pd.Series, threshold: float = 0.5) -> bool:
    """Safely checks if a column predominantly contains list-like objects.

    Args:
        column (pd.Series):
            The pandas Series (column) to check. If not a Series, returns False and logs an error.
        threshold (float, optional):
            0.0 < threshold <= 1.0. Float representation of the percentage.
            Default = 0.5 (50%). If out of bounds, returns False and logs an error.

    Returns:
        bool:
            True if the column contains mostly list-like objects, False otherwise.

    """

    if not isinstance(column, pd.Series):
        self.logger.error(
            "Expected pandas series, but got -> %s",
            str(type(column)),
        )
        return False
    if not 0.0 < threshold <= 1.0:
        self.logger.error(
            "Threshold must be between 0.0 and 1.0, but got -> %s",
            str(threshold),
        )
        return False

    # Drop null values (NaN or None) and check types of remaining values
    non_null_values = column.dropna()
    list_count = non_null_values.apply(lambda x: isinstance(x, list)).sum()

    # If more than threshold % of non-null values are lists, return True.
    # Else return False.
    return list_count / len(non_null_values) > threshold if len(non_null_values) > 0 else False

is_string_column

is_string_column(column: Series) -> bool

Determine if a Pandas series contains only string values, ignoring NaN values.

Parameters:

Name	Type	Description	Default
column	Series	The Pandas Series to check.	required

Returns:

Name	Type	Description
bool	bool	True if all non-NaN values in the column are strings, False otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def is_string_column(self, column: pd.Series) -> bool:
    """Determine if a Pandas series contains only string values, ignoring NaN values.

    Args:
        column (pd.Series):
            The Pandas Series to check.

    Returns:
        bool:
            True if all non-NaN values in the column are strings, False otherwise.

    """

    # Drop NaN values and check if remaining values are strings
    return column.dropna().map(lambda x: isinstance(x, str)).all()

cleanse

cleanse(cleansings: list) -> None

Cleanse data with regular expressions and upper/lower case conversions.

Parameters:

Name	Type	Description	Default
cleansings	list	List of cleansing dictionaries. Each dictionary has these fields: * field (str, required): name of a column in the data frame * replacements (dict, optional, default = {}): regex patterns and replacements * upper (bool, optional, default = False): change the value to uppercase * lower (bool, optional, default = False): change the value to lowercase * capitalize (bool, optional, default = False) - first character upper case, rest lower-case * title (bool, optional, default = False) - first character of each word upper case * length (int, optional, default = 0): truncate to max length	required

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def cleanse(self, cleansings: list) -> None:
    """Cleanse data with regular expressions and upper/lower case conversions.

    Args:
        cleansings (list):
            List of cleansing dictionaries. Each dictionary has
            these fields:
            * field (str, required): name of a column in the data frame
            * replacements (dict, optional, default = {}): regex patterns and replacements
            * upper (bool, optional, default = False): change the value to uppercase
            * lower (bool, optional, default = False): change the value to lowercase
            * capitalize (bool, optional, default = False) - first character upper case, rest lower-case
            * title (bool, optional, default = False) - first character of each word upper case
            * length (int, optional, default = 0): truncate to max length

    """

    # Iterate over each cleansing entry in the list
    for cleansing in cleansings:
        column = cleansing.get("field", "")
        if not column:
            self.logger.error("Cleansing entry is missing the 'field' key. Skipping...")
            continue

        # Read the cleansing parameters:
        replacements = cleansing.get("replacements", {})
        upper = cleansing.get("upper", False)
        lower = cleansing.get("lower", False)
        capitalize = cleansing.get("capitalize", False)
        title = cleansing.get("title", False)
        length = cleansing.get("length", 0)

        # Handle dict columns - we expect the column name to separate
        # main field from sub field using a dot syntax (e.g., "column.subfield")
        if "." in column:
            column, dict_key = column.split(".")
            if column not in self._df.columns:
                self.logger.error(
                    "Cannot cleanse column -> '%s'. It does not exist in the data frame! Data frame has these columns -> %s",
                    column,
                    str(self._df.columns),
                )
                continue
            # Apply cleansing to dictionary values in the main column
            self.logger.debug(
                "Cleansing for column -> '%s' has a subfield -> '%s' configured. Do cleansing for dictionary items with key -> '%s'...",
                column,
                dict_key,
                dict_key,
            )
            self._df[column] = self._df[column].apply(
                lambda x, dict_key=dict_key, replacements=replacements, upper=upper, lower=lower, capitalize=capitalize, title=title, length=length: (
                    self._cleanse_subfield(
                        data=x,
                        dict_key=dict_key,
                        replacements=replacements,
                        upper=upper,
                        lower=lower,
                        capitalize=capitalize,
                        title=title,
                        length=length,
                    )
                ),
            )
        # end if "." in column
        else:  # the else case handles strings and list columns
            if column not in self._df.columns:
                self.logger.error(
                    "Cannot cleanse column -> '%s'. It does not exist in the data frame! Data frame has these columns -> %s",
                    column,
                    str(self._df.columns),
                )
                continue

            # Handle string columns:
            if self.is_string_column(self._df[column]):
                # Apply cleansing operations on string column
                self.logger.debug(
                    "Column -> '%s' has string values. Do cleansing for string values...",
                    column,
                )
                self._df[column] = self._df[column].apply(
                    lambda x, replacements=replacements, upper=upper, lower=lower, capitalize=capitalize, title=title, length=length: (
                        self._apply_string_cleansing(
                            value=x,
                            replacements=replacements,
                            upper=upper,
                            lower=lower,
                            capitalize=capitalize,
                            title=title,
                            length=length,
                        )
                        if isinstance(x, str)
                        else x
                    ),
                )

            # Handle list columns:
            elif self.is_list_column(self._df[column]):
                # Handle list-like columns for this we iterate over each list item
                # and apply the cleansing by calling _apply_string_cleansing() for item:
                self.logger.debug(
                    "Column -> '%s' has list values. Do cleansing for each list item...",
                    column,
                )
                self._df[column] = self._df[column].apply(
                    lambda x, replacements=replacements, upper=upper, lower=lower, capitalize=capitalize, title=title, length=length: (
                        [
                            (
                                self._apply_string_cleansing(
                                    value=item,
                                    replacements=replacements,
                                    upper=upper,
                                    lower=lower,
                                    capitalize=capitalize,
                                    title=title,
                                    length=length,
                                )
                                if isinstance(
                                    item,
                                    str,
                                )  # we just change string list items
                                else item
                            )
                            for item in x
                        ]
                        if isinstance(x, list)
                        else x
                    ),
                )

            else:
                self.logger.error(
                    "Column -> '%s' is not a string, list, or dict-like column. Skipping cleansing...",
                    column,
                )

filter

filter(
    conditions: list,
    inplace: bool = True,
    reset_index: bool = True,
) -> DataFrame | None

Filter the data frame based on (multiple) conditions.

Parameters:

Name	Type	Description	Default
conditions	list	Conditions are a list of dictionaries with these items: * field (str): The name of a column in the data frame * value (str or list): Expected value (filter criterium). Alternatively the key 'values' can be used. If it is a list then one of the list elements must match the field value (OR) * equal (bool): Whether to test for equal or non-equal. If not specified equal is treated as True. * regex (bool): This flag controls if the value is interpreted as a regular expression. If there is no regex item in the dictionary then the default is False (= values is NOT regex). * enabled (bool): True or False. The filter is only applied if 'enabled = True'. If not specified the condition is treated as enabled. If there are multiple conditions in the list each has to evaluate to True (AND)	required
inplace	bool	Defines if the self._df is modified (inplace) or just a new data frame is returned. Defaults to True.	True
reset_index	bool	Filter removes rows. If reset_index = True then the numbering of the index is newly calculated (only done if inplace is True). Defaults to True.	True

Returns:

Type	Description
DataFrame | None	pd.DataFrame | None: A new data frame or pointer to self._df (depending on the value of 'inplace'). None in case of an error.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def filter(
    self,
    conditions: list,
    inplace: bool = True,
    reset_index: bool = True,
) -> pd.DataFrame | None:
    """Filter the data frame based on (multiple) conditions.

    Args:
        conditions (list):
            Conditions are a list of dictionaries with these items:
            * field (str): The name of a column in the data frame
            * value (str or list):
                Expected value (filter criterium). Alternatively the key 'values' can be used.
                If it is a list then one of the list elements must match the field value (OR)
            * equal (bool):
                Whether to test for equal or non-equal. If not specified equal is treated as True.
            * regex (bool):
                This flag controls if the value is interpreted as a
                regular expression. If there is no regex item in the
                dictionary then the default is False (= values is NOT regex).
            * enabled (bool):
                True or False. The filter is only applied if 'enabled = True'.
                If not specified the condition is treated as enabled.
            If there are multiple conditions in the list each has to evaluate to True (AND)
        inplace (bool, optional):
            Defines if the self._df is modified (inplace) or just
            a new data frame is returned. Defaults to True.
        reset_index (bool, optional):
            Filter removes rows. If reset_index = True then the numbering
            of the index is newly calculated (only done if inplace is True).
            Defaults to True.

    Returns:
        pd.DataFrame | None:
            A new data frame or pointer to self._df (depending on the value of 'inplace').
            None in case of an error.

    """

    if self._df is None:
        self.logger.error("Data frame is not initialized.")
        return None

    if self._df.empty:
        self.logger.error("Data frame is empty.")
        return None

    # First filtered_df is the full data frame.
    # Then it is subsequentially reduced by each condition
    # at the end it is just those rows that match all conditions.
    filtered_df = self._df if inplace else self._df.copy()

    def list_matches(row: list, values: list) -> bool:
        """Check if any item in the 'values' list is present in the given 'row' list.

        Args:
            row (list):
                A list of items from the data frame column.
            values (list):
                A list of values to check for in the 'row'.

        Returns:
            bool:
                True if any item in 'values' is found in 'row', otherwise False.

        """

        return any(item in values for item in row)

    def dict_matches(row: dict, key: str, values: list) -> bool:
        """Check if the value for the dictionary 'key' is in 'values'.

        Args:
            row (dict):
                A dictionary from the data frame column.
            key (str):
                The key to lookup in the dictionary.
            values (list):
                A list of values to check for in the 'row'.

        Returns:
            bool:
                True, if the value for the dictionary key is in 'values', otherwise False.

        """

        if not row or key not in row:
            return False

        return row[key] in values

    # We traverse a list of conditions. Each condition must evaluate to True
    # otherwise the current workspace or document (i.e. the data set for these objects)
    # will be skipped.
    for condition in conditions:
        # Check if the condition is enabled. If 'enabled' is not
        # in the condition dict then we assume it is enabled.
        if not condition.get("enabled", True):
            continue
        field = condition.get("field", None)
        if not field:
            self.logger.error(
                "Missing value for filter condition 'field' in payload!",
            )
            continue
        if "." in field:
            field, sub = field.split(".", 1)
        else:
            sub = None

        if field not in self._df.columns:
            self.logger.warning(
                "Filter condition field -> '%s' does not exist as column in the data frame! Data frame has these columns -> %s",
                field,
                str(self._df.columns),
            )
            continue  # Skip filtering for columns not present in data frame

        regex = condition.get("regex", False)
        # We need the column to be of type string if we want to use regular expressions
        # so if the column is not yet a string we convert the column to string:
        if regex and filtered_df[field].dtype != "object":
            # Change type of column to string:
            filtered_df[field] = filtered_df[field].astype(str)
            filtered_df[field] = filtered_df[field].fillna("")

        value = condition.get("value", None)
        if value is None:
            # Support alternative syntax using plural.
            value = condition.get("values", None)
        if value is None:
            self.logger.error(
                "Missing filter value(s) for filter condition field -> '%s'!",
                field,
            )
            continue

        # if a single string is passed as value we put
        # it into an 1-item list to simplify the following code:
        if not isinstance(value, list):
            value = [value]

        # If all value items in the value list of the condition are strings
        # then we want the column to filter on also to be of type string
        # and we cast the column to string type:
        if all(isinstance(v, str) for v in value):
            # Change type of column to string:
            #                filtered_df[field] = filtered_df[field].astype(str)
            #                filtered_df[field] = filtered_df[field].fillna("").astype(str)
            #                filtered_df[field] = filtered_df[field].fillna("")

            # When inplace == True, filtered_df is just a reference to self._df.
            # Using .loc[:, field] ensures that Pandas updates the column correctly in self._df.
            # When inplace == False, filtered_df is a full copy (self._df.copy() above),
            # so modifications remain in filtered_df.
            # .loc[:, field] ensures no SettingWithCopyWarning, since filtered_df is now a separate DataFrame.
            filtered_df.loc[:, field] = filtered_df[field].fillna("").astype(str)

        self.logger.info(
            "Data frame has %s row(s) and %s column(s) before filter -> %s has been applied.",
            str(filtered_df.shape[0]),
            str(filtered_df.shape[1]),
            str(condition),
        )

        # Check if the column is boolean
        if pd.api.types.is_bool_dtype(filtered_df[field]):
            # Convert string representations of booleans to actual booleans
            value = [v.lower() in ["true", "1"] if isinstance(v, str) else bool(v) for v in value]

        # Do we want to test for equalitiy or non-equality?
        # For lists equality means: value is in the list
        # For lists non-equality means: value is NOT in the list
        test_for_equal = condition.get("equal", True)

        # Check if the column contains only lists (every non-empty element in the column is a list).
        # `filtered_df[field]`: Access the column with the name specified in 'field'.
        # `.dropna()`: Drop None or NaN rows for the test.
        # `.apply(lambda x: isinstance(x, list))`: For each element in the column, check if it is a list.
        # `.all()`: Ensure that all elements in the column satisfy the condition of being a list.
        if filtered_df[field].dropna().apply(lambda x: isinstance(x, list)).all():
            if not test_for_equal:
                filtered_df = filtered_df[~filtered_df[field].apply(list_matches, values=value)]
            else:
                filtered_df = filtered_df[filtered_df[field].apply(list_matches, values=value)]
        # Check if the column contains only dictionaries (every non-empty element in the column is a dict).
        # `filtered_df[field]`: Access the column with the name specified in 'field'.
        # `.dropna()`: Drop None or NaN rows for the test.
        # `.apply(lambda x: isinstance(x, dict))`: For each element in the column, check if it is a dict.
        # `.all()`: Ensure that all elements in the column satisfy the condition of being a dictionary.
        elif filtered_df[field].dropna().apply(lambda x: isinstance(x, dict)).all():
            if not sub:
                self.logger.error(
                    "Filtering on dictionary values need a key. This needs to be provided with 'field.key' syntax!",
                )
                continue
            if not test_for_equal:
                filtered_df = filtered_df[~filtered_df[field].apply(dict_matches, key=sub, values=value)]
            else:
                filtered_df = filtered_df[filtered_df[field].apply(dict_matches, key=sub, values=value)]
        # Check if the column has boolean values:
        elif pd.api.types.is_bool_dtype(filtered_df[field]):
            # For a boolean filter we can drop NA values:
            filtered_df = filtered_df.dropna(subset=[field])
            if not test_for_equal:
                filtered_df = filtered_df[~filtered_df[field].isin(value)]
            else:
                filtered_df = filtered_df[filtered_df[field].isin(value)]
        elif regex:
            # Create a pure boolean pd.Series as a filter criterium:
            regex_condition = filtered_df[field].str.contains(
                "|".join(value),
                regex=True,
                na=False,
            )
            # Apply the boolean pd.Series named 'regex_condition' as
            # a filter - either non-negated or negated (using ~):
            filtered_df = filtered_df[~regex_condition] if not test_for_equal else filtered_df[regex_condition]
        else:
            # If it is a column with string values we strip leading/trailing spaces
            if pd.api.types.is_string_dtype(filtered_df[field]):
                filtered_df[field] = filtered_df[field].str.strip()
            if not test_for_equal:
                filtered_df = filtered_df[~filtered_df[field].isin(value)]
            else:
                filtered_df = filtered_df[filtered_df[field].isin(value)]

        self.logger.info(
            "Data frame has %s row(s) and %s column(s) after filter -> %s has been applied.",
            str(filtered_df.shape[0]),
            str(filtered_df.shape[1]),
            str(condition),
        )
    # end for condition

    if inplace:
        self._df = filtered_df

        if reset_index:
            self._df.reset_index(inplace=True, drop=True)

    return filtered_df

fill_na_in_column

fill_na_in_column(
    column_name: str, default_value: str | int
) -> None

Replace NA values in a column with a defined new default value.

Parameters:

Name	Type	Description	Default
column_name	str	The name of the column in the data frame.	required
default_value	str | int	The value to replace NA with.	required

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def fill_na_in_column(self, column_name: str, default_value: str | int) -> None:
    """Replace NA values in a column with a defined new default value.

    Args:
        column_name (str):
            The name of the column in the data frame.
        default_value (str | int):
            The value to replace NA with.

    """

    if column_name in self._df.columns:
        self._df[column_name] = self._df[column_name].fillna(value=default_value)
    else:
        self.logger.error(
            "Cannot replace NA values as column -> '%s' does not exist in the data frame! Available columns -> %s",
            column_name,
            str(self._df.columns),
        )

fill_forward

fill_forward(inplace: bool) -> DataFrame

Fill the missing cells appropriately by carrying forward the values from the previous rows where necessary.

This has applications if a hierarchy is represented by nested cells e.g. in an Excel sheet.

Parameters:

Name	Type	Description	Default
inplace	bool	Should the modification happen inplace or not.	required

Returns:

Type	Description
DataFrame	pd.DataFrame: The resulting data frame. Note that pandas returns None from ffill() if inplace is True.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def fill_forward(self, inplace: bool) -> pd.DataFrame:
    """Fill the missing cells appropriately by carrying forward the values from the previous rows where necessary.

    This has applications if a hierarchy is represented by
    nested cells e.g. in an Excel sheet.

    Args:
        inplace (bool):
            Should the modification happen inplace or not.

    Returns:
        pd.DataFrame:
            The resulting data frame. Note that pandas returns None from ffill()
            if inplace is True.

    """

    # To convert an Excel representation of a folder structure with nested
    # columns into a format appropriate for Pandas,
    # where all cells should be filled
    df_filled = self._df.ffill(inplace=inplace)

    return df_filled

lookup_value

lookup_value(
    lookup_column: str,
    lookup_value: str,
    separator: str = "|",
    single_row: bool = True,
) -> Series | DataFrame | None

Lookup row(s) that includes a lookup value in the value of a given column.

Parameters:

Name	Type	Description	Default
lookup_column	str	The name of the column to search in.	required
lookup_value	str	The value to search for.	required
separator	str	The string list delimiter / separator. The pipe symbol | is the default as it is unlikely to appear in a normal string (other than a plain comma). The separator is NOT looked for in the lookup_value but in the column that is given by lookup_column!	'|'
single_row	bool	This defines if we just return the first matching row if multiple matching rows are found. Default is True (= single row).	True

Returns:

Type	Description
Series | DataFrame | None	pd.Series | pd.DataFrame | None: Data frame (multiple rows) or Series (row) that matches the lookup value. None if no match was found.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def lookup_value(
    self,
    lookup_column: str,
    lookup_value: str,
    separator: str = "|",
    single_row: bool = True,
) -> pd.Series | pd.DataFrame | None:
    """Lookup row(s) that includes a lookup value in the value of a given column.

    Args:
        lookup_column (str):
            The name of the column to search in.
        lookup_value (str):
            The value to search for.
        separator (str, optional):
            The string list delimiter / separator. The pipe symbol | is the default
            as it is unlikely to appear in a normal string (other than a plain comma).
            The separator is NOT looked for in the lookup_value but in the column that
            is given by lookup_column!
        single_row (bool, optional):
            This defines if we just return the first matching row if multiple matching rows
            are found. Default is True (= single row).

    Returns:
        pd.Series | pd.DataFrame | None:
            Data frame (multiple rows) or Series (row) that matches the lookup value.
            None if no match was found.

    """

    # Use the `apply` function to filter rows where the lookup value matches a
    # whole item in the separator-divided list:
    def match_lookup_value(string_list: str | None) -> bool:
        """Check if the lookup value is in a string list.

        For this the string list is converted to a python
        list. A separator is used for the splitting.

        Args:
            string_list (str):
                Delimiter-separated string list like "a, b, c" or "a | b | c"

        Returns:
            bool:
                True if lookup_value is equal to one of the delimiter-separated terms.

        """

        if pd.isna(string_list):  # Handle None/NaN safely
            return False

        # Ensure that the string is a string
        string_list = str(string_list)

        return lookup_value in [item.strip() for item in string_list.split(separator)]

    # end method definition

    if self._df is None:
        return None

    df = self._df

    if lookup_column not in self._df.columns:
        self.logger.error(
            "Cannot lookup value in column -> '%s'. Column does not exist in the data frame! Data frame has these columns -> %s",
            lookup_column,
            str(self._df.columns),
        )
        return None

    # Fill NaN or None values in the lookup column with empty strings
    # df[lookup_column] = df[lookup_column].fillna("")

    # Use the `apply` function to filter rows where the lookup value is in row cell
    # of column given by lookup_column. match_lookup_value() is called with
    # the content of the individual cell contents:
    matched_rows = df[df[lookup_column].apply(match_lookup_value)]

    # If nothing was found we return None:
    if matched_rows.empty:
        return None

    # If it is OK to have multiple matches (= multiple rows = pd.DataFrame).
    # We can just return the matched_rows now which should be a pd.DataFrame:
    if not single_row:
        return matched_rows

    # Check if more than one row matches, and log a warning if so
    if len(matched_rows) > 1:
        self.logger.warning(
            "More than one match found for lookup value -> '%s' in column -> '%s'. Returning the first match.",
            lookup_value,
            lookup_column,
        )

    # Return the first matched row, if any
    return matched_rows.iloc[0]

set_value

set_value(
    column: str, value, condition: Series | None = None
) -> None

Set the value in the data frame based on a condition.

Parameters:

Name	Type	Description	Default
column	str	The name of the column.	required
value	Any	The value to set for those rows that fulfill the condition.	required
condition	Series | None	This should be a boolean Series where each element is True or False, representing rows in the data frame that meet a certain condition. If None is provided then ALL rows get the 'value' in the given column.	None

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def set_value(self, column: str, value, condition: pd.Series | None = None) -> None:  # noqa: ANN001
    """Set the value in the data frame based on a condition.

    Args:
        column (str):
            The name of the column.
        value (Any):
            The value to set for those rows that fulfill the condition.
        condition (pd.Series | None, optional):
            This should be a boolean Series where each element is True or False,
            representing rows in the data frame that meet a certain condition.
            If None is provided then ALL rows get the 'value' in the given
            column.

    """

    if condition is None:
        self._df[column] = value  # Set value unconditionally
    else:
        self._df.loc[condition, column] = value  # Set value based on condition

add_column

add_column(
    new_column: str,
    data_type: str = "string",
    source_column: str = "",
    reg_exp: str = "",
    prefix: str = "",
    suffix: str = "",
    length: int | None = None,
    group_chars: int | None = None,
    group_separator: str = ".",
    group_remove_leading_zero: bool = True,
) -> bool

Add additional column to the data frame.

Parameters:

Name	Type	Description	Default
new_column	str	The name of the column to add.	required
data_type	str	The data type of the new column. This is only used if no source column is given (i.e. an empty column is added). Defaults to "string".	'string'
source_column	str	The name of the source column. If empty, an empty new column is added. Defaults to "".	''
reg_exp	str	A regular expression to apply on the content of the source column (using str.extract, so it must contain exactly one capture group). Defaults to "".	''
prefix	str	Prefix to add in front of the value. Defaults to "".	''
suffix	str	Suffix to add at the end of the value. Defaults to "".	''
length	int | None	Length to reduce to. Defaults to None (= unlimited).	None
group_chars	int | None	Group the resulting string in characters of group_chars. Defaults to None. Usable e.g. for thousand separator "."	None
group_separator	str	Separator string for the grouping. Defaults to ".".	'.'
group_remove_leading_zero	bool	Remove leading zeros from the groups. Defaults to True.	True

Returns:

Name	Type	Description
bool	bool	True = Success, False = Failure

Note

self._df is modified in place.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def add_column(
    self,
    new_column: str,
    data_type: str = "string",
    source_column: str = "",
    reg_exp: str = "",
    prefix: str = "",
    suffix: str = "",
    length: int | None = None,
    group_chars: int | None = None,
    group_separator: str = ".",
    group_remove_leading_zero: bool = True,
) -> bool:
    """Add additional column to the data frame.

    Args:
        new_column (str):
            The name of the column to add.
        data_type (str, optional):
            The data type of the new column. This is only used if no source column
            is given (i.e. an empty column is added). Defaults to "string".
        source_column (str, optional):
            The name of the source column. If empty, an empty new column is added.
            Defaults to "".
        reg_exp (str, optional):
            A regular expression to apply on the content of the source column
            (using `str.extract`, so it must contain exactly one capture group).
            Defaults to "".
        prefix (str, optional):
            Prefix to add in front of the value. Defaults to "".
        suffix (str, optional):
            Suffix to add at the end of the value. Defaults to "".
        length (int | None, optional):
            Length to reduce to. Defaults to None (= unlimited).
        group_chars (int | None, optional):
            Group the resulting string in characters of group_chars. Defaults to None.
            Usable e.g. for thousand separator "."
        group_separator (str, optional):
            Separator string for the grouping. Defaults to ".".
        group_remove_leading_zero (bool, optional):
            Remove leading zeros from the groups. Defaults to True.

    Returns:
        bool:
            True = Success, False = Failure

    Note:
        self._df is modified in place.

    """

    if self._df is None:
        self.logger.error("Data frame is not initialized. Cannot add column -> %s!", new_column)
        return False

    # Check that the new column does not yet exist
    if new_column in self._df.columns:
        self.logger.error(
            "New column -> '%s' does already exist in data frame! Cannot add it. Data frame has these columns -> %s",
            new_column,
            str(self._df.columns),
        )
        return False

    # first we handle the very simple case to not have a source column but just add an empty new column.
    # It is important to add the index parameter as Series assignment is index-aligned, not positional:
    if not source_column:
        self._df[new_column] = pd.Series(index=self._df.index, dtype=data_type)
        return True

    # Check if the source column exists
    if source_column not in self._df.columns:
        self.logger.error(
            "Source column -> '%s' does not exist as column in data frame! Data frame has these columns -> %s",
            source_column,
            str(self._df.columns),
        )
        return False

    # Validate the regex pattern
    try:
        re.compile(reg_exp)  # Check if the pattern is a valid regex
    except re.error:
        self.logger.error(
            "Invalid regular expression -> %s. Cannot extract data for new column -> '%s'!",
            reg_exp,
            new_column,
        )
        return False

    # Ensure the source column is of type string (convert it, if necessary)
    if self._df[source_column].dtype != "object":
        self._df[source_column] = self._df[source_column].astype(str)

    # Use str.extract to apply the regular expression to the source column
    # and then assign this modified column to the variable "extracted":
    extracted = self._df[source_column].str.extract(pat=reg_exp, expand=False)

    # Limit the result to the specified length
    if length is not None:
        extracted = extracted.str[:length]

    if group_chars is not None:

        def process_grouping(x) -> str | None:  # noqa: ANN001
            if pd.isna(x):
                return None
            # Split into groups
            groups = [x[i : i + group_chars] for i in range(0, len(x), group_chars)]
            if group_remove_leading_zero:
                # Remove leading zeros from each group
                groups = [group.lstrip("0") or "0" for group in groups]
            # Join groups with separator
            return group_separator.join(groups)

        extracted = extracted.apply(process_grouping)

    # Add prefix and suffix
    if prefix or suffix:
        extracted = prefix + extracted.astype(str) + suffix

    self._df[new_column] = extracted

    return True

convert_to_lists

convert_to_lists(
    columns: list, delimiter: str = ","
) -> None

Intelligently convert string values to list values, in defined data frame columns.

The delimiter to separate values in the string value can be configured. The method is ignoring delimiters that are inside quotes.

Parameters:

Name	Type	Description	Default
columns	list	The name of the columns whose values should be converted to lists.	required
delimiter	str	Character that delimits list items. Defaults to ",".	','

Returns:

Type	Description
None	None. self._df is modified in place.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def convert_to_lists(self, columns: list, delimiter: str = ",") -> None:
    """Intelligently convert string values to list values, in defined data frame columns.

    The delimiter to separate values in the string value can be configured.
    The method is ignoring delimiters that are inside quotes.

    Args:
        columns (list):
            The name of the columns whose values should be converted to lists.
        delimiter (str, optional):
            Character that delimits list items. Defaults to ",".

    Returns:
        None. self._df is modified in place.

    """

    # Regex to split by the delimiter, ignoring those inside quotes or double quotes
    def split_string_ignoring_quotes(s: str, delimiter: str) -> list:
        """Split a string into a list at positions that have a delimiter character.

        Args:
            s (str): the string to split
            delimiter (str): The single character that is used for splitting.

        Returns:
            A list of split values.

        """

        # Escaping the delimiter in case it's a special regex character
        delimiter = re.escape(delimiter)
        # Match quoted strings and unquoted delimiters separately
        pattern = rf'(?:"[^"]*"|\'[^\']*\'|[^{delimiter}]+)'
        return re.findall(pattern, s)

    for col in columns:
        self._df[col] = self._df[col].apply(
            lambda x: split_string_ignoring_quotes(x, delimiter) if isinstance(x, str) and delimiter in x else x,
        )

add_column_concat

add_column_concat(
    source_columns: list,
    new_column: str,
    concat_char: str = "",
    upper: bool = False,
    lower: bool = False,
    capitalize: bool = False,
    title: bool = False,
) -> bool

Add a column as a concatenation of the values of multiple source columns.

Parameters:

Name	Type	Description	Default
source_columns	list	The names of the columns whose values are concatenated.	required
new_column	str	The name of the new column.	required
concat_char	str	Character to insert between the concatenated values. Default is "".	''
upper	bool	Convert result to uppercase if True.	False
lower	bool	Convert result to lowercase if True.	False
capitalize	bool	Capitalize the result if True.	False
title	bool	Convert result to title case if True.	False

Returns:

Name	Type	Description
bool	bool	True = Success, False = Failure

Note

self._df is modified in place.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def add_column_concat(
    self,
    source_columns: list,
    new_column: str,
    concat_char: str = "",
    upper: bool = False,
    lower: bool = False,
    capitalize: bool = False,
    title: bool = False,
) -> bool:
    """Add a column as a concatenation of the values of multiple source columns.

    Args:
        source_columns (list):
            The names of the columns whose values are concatenated.
        new_column (str):
            The name of the new column.
        concat_char (str, optional):
            Character to insert between the concatenated values. Default is "".
        upper (bool, optional):
            Convert result to uppercase if True.
        lower (bool, optional):
            Convert result to lowercase if True.
        capitalize (bool, optional):
            Capitalize the result if True.
        title (bool, optional):
            Convert result to title case if True.

    Returns:
        bool:
            True = Success, False = Failure

    Note:
        self._df is modified in place.

    """

    def concatenate(row: pd.Series) -> str:
        # Comprehension to create a list from all source column values:
        concatenated = concat_char.join(
            [str(row[col]) for col in source_columns if pd.notna(row[col])],
        )

        # Apply case transformations based on parameters
        if upper:
            concatenated = concatenated.upper()
        elif lower:
            concatenated = concatenated.lower()
        elif capitalize:
            concatenated = concatenated.capitalize()
        elif title:
            concatenated = concatenated.title()

        return concatenated

    # end method definition

    #
    # Validations:
    #

    # Check the data frame is initialized:
    if self._df is None:
        self.logger.error(
            "Data frame is not initialized. Cannot add a column -> '%s' via concatenation!", new_column
        )
        return False
    # Check we only do one case transformation as they are mutually exclusive:
    if sum([upper, lower, capitalize, title]) > 1:
        self.logger.warning("Only one case transformation can be applied for added data frame column.")
    # Check that the new column does not yet exist:
    if new_column in self._df.columns:
        self.logger.error(
            "New column -> '%s' does already exist in data frame! Cannot add it by concatenation of -> %s. Data frame has these columns -> %s",
            new_column,
            str(source_columns),
            str(self._df.columns),
        )
        return False
    # Check source columns are valid:
    if not isinstance(source_columns, (list, tuple)) or not source_columns:
        self.logger.error("Source columns must be a non-empty list of column names!")
        return False
    missing = [c for c in source_columns if c not in self._df.columns]
    if missing:
        self.logger.error(
            "Missing source columns -> %s. Cannot concat it to values of new column -> '%s'!",
            str(missing),
            new_column,
        )
        return False

    #
    # Execute the transformation:
    #
    self._df[new_column] = self._df.apply(concatenate, axis=1)

    return True

add_column_list

add_column_list(
    source_columns: list, new_column: str
) -> bool

Add a column with list objects.

The list items are taken from a list of source columns (row by row).

Parameters:

Name	Type	Description	Default
source_columns	list	The column names the list values are taken from.	required
new_column	str	The name of the new column.	required

Returns:

Name	Type	Description
bool	bool	True = Success, False = Failure

Note

self._df is modified in place.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def add_column_list(self, source_columns: list, new_column: str) -> bool:
    """Add a column with list objects.

    The list items are taken from a list of source columns (row by row).

    Args:
        source_columns (list):
            The column names the list values are taken from.
        new_column (str):
            The name of the new column.

    Returns:
        bool:
            True = Success, False = Failure

    Note:
        self._df is modified in place.

    """

    def create_list(row: pd.Series) -> list:
        # Comprehension to create a list from all source column values:
        return [row[col] for col in source_columns]

    #
    # Validations:
    #

    # Check the data frame is initialized:
    if self._df is None:
        self.logger.error("Data frame is not initialized. Cannot add new list column -> '%s'!", new_column)
        return False
    # Check that the new column does not yet exist:
    if new_column in self._df.columns:
        self.logger.error("Column -> '%s' already exists in the data frame. Cannot add it!", new_column)
        return False
    # Check source columns are valid:
    if not isinstance(source_columns, (list, tuple)) or not source_columns:
        self.logger.error("Source columns must be a non-empty list of column names! Cannot add new list column.")
        return False
    missing = [c for c in source_columns if c not in self._df.columns]
    if missing:
        self.logger.error(
            "Missing source columns -> %s. Cannot transform its values to a new list column -> '%s'!",
            str(missing),
            new_column,
        )
        return False

    #
    # Execute the transformation:
    #
    self._df[new_column] = self._df.apply(create_list, axis=1)

    return True

add_column_table

add_column_table(
    source_columns: list,
    new_column: str,
    delimiter: str = ",",
) -> bool

Add a column with tabular objects (list of dictionaries).

The source columns should include lists. The resulting dictionary keys are the column names for the source columns.

Example (["X", "Y"] are the source_columns, "Table" is the new_column): X[1] = [1, 2, 3] # row 1 Y[1] = ["A", "B", "C"] # row 1 X[2] = [4, 5, 6] # row 2 Y[2] = ["D", "E", "F"] # row 2

Table[1] = [ { "X": "1" "Y": "A" }, { "X": "2" "Y": "B" }, { "X": "3" "Y": "C" } ] Table[2] = [ { "X": "4" "Y": "D" }, { "X": "5" "Y": "E" }, { "X": "6" "Y": "F" } ]

Parameters:

Name	Type	Description	Default
source_columns	list	The column names the list values are taken from.	required
new_column	str	The name of the new column.	required
delimiter	str	Character that delimits list items. Defaults to ",".	','

Returns:

Name	Type	Description
bool	bool	True = Success, False = Failure

Note

self._df is modified in place.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def add_column_table(
    self,
    source_columns: list,
    new_column: str,
    delimiter: str = ",",
) -> bool:
    """Add a column with tabular objects (list of dictionaries).

    The source columns should include lists. The resulting dictionary
    keys are the column names for the source columns.

    Example (["X", "Y"] are the source_columns, "Table" is the new_column):
    X[1] = [1, 2, 3]         # row 1
    Y[1] = ["A", "B", "C"]   # row 1
    X[2] = [4, 5, 6]         # row 2
    Y[2] = ["D", "E", "F"]   # row 2

    Table[1] = [
        {
            "X": "1"
            "Y": "A"
        },
        {
            "X": "2"
            "Y": "B"
        },
        {
            "X": "3"
            "Y": "C"
        }
    ]
    Table[2] = [
        {
            "X": "4"
            "Y": "D"
        },
        {
            "X": "5"
            "Y": "E"
        },
        {
            "X": "6"
            "Y": "F"
        }
    ]

    Args:
        source_columns (list):
            The column names the list values are taken from.
        new_column (str):
            The name of the new column.
        delimiter (str, optional):
            Character that delimits list items. Defaults to ",".

    Returns:
        bool:
            True = Success, False = Failure

    Note:
        self._df is modified in place.

    """

    # Sub-method to pad lists to the desired length
    def pad_list(lst: list, max_len: int) -> list:
        """Pad lists to the same length. None is used as the filler.

        Args:
            lst (list):
                List to pad.
            max_len (int):
                Desired length of the list.

        Returns:
            list:
                The padded list.

        """

        return lst + [None] * (max_len - len(lst))

    # end sub-method

    def create_table(row: pd.Series) -> list:
        """Create a list of dictionaries representing the table value.

        Args:
            row (pd.Series):
                Current row to process.

        Returns:
            list:
                List of dictionaries representing the table value.

        """

        # Step 1: Determine maximum length across columns
        max_len = max(len(row[col]) if isinstance(row[col], list) else 1 for col in source_columns)

        # Step 2: Pad lists to the maximum length, leave scalar values as they are, don't change source columns:
        table_values = {}  # list of values padded to max_len. We use a separate variable to not change source columns.
        for col in source_columns:
            val = row[col]
            if isinstance(val, list):
                table_values[col] = pad_list(val, max_len)
            elif not pd.isna(val):
                table_values[col] = [
                    val,
                ] * max_len  # Repeat scalar value to match the max length
            else:
                table_values[col] = [None] * max_len  # fill missing values

        # Step 3: Create a list of dictionaries (table) for each row:
        table = [{col: table_values[col][i] for col in source_columns} for i in range(max_len)]

        return table

    # end sub-method

    #
    # Validations:
    #

    # Check the data frame is initialized:
    if self._df is None:
        self.logger.error("Data frame is not initialized. Cannot add new table column -> '%s'!", new_column)
        return False
    # Check that the new column does not yet exist:
    if new_column in self._df.columns:
        self.logger.error("Column -> '%s' already exists in the data frame. Cannot add it!", new_column)
        return False
    # Check source columns are valid:
    if not isinstance(source_columns, (list, tuple)) or not source_columns:
        self.logger.error("Source columns must be a non-empty list of column names!")
        return False
    missing = [c for c in source_columns if c not in self._df.columns]
    if missing:
        self.logger.error(
            "Missing source columns -> %s. Cannot transform it to table values in new column -> '%s'!",
            str(missing),
            new_column,
        )
        return False

    #
    # Execute the transformations:
    #

    # Call the convert_to_lists method to ensure the columns are converted
    self.convert_to_lists(columns=source_columns, delimiter=delimiter)

    # Apply the function to create a new column with table values:
    self._df[new_column] = self._df.apply(create_table, axis=1)

    return True

drop_row

drop_row(index: int | str) -> None

Drop a single row from the DataFrame by its index.

Parameters:

Name	Type	Description	Default
index	int | str	The index value of the row to remove. Can be an integer or string depending on the DataFrame index type.	required

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def drop_row(self, index: int | str) -> None:
    """Drop a single row from the DataFrame by its index.

    Args:
        index (int | str):
            The index value of the row to remove. Can be an integer or string
            depending on the DataFrame index type.

    """

    if index in self._df.index:
        self._df.drop(index=index, inplace=True)

drop_rows

drop_rows(mask: Series) -> None

Drop rows from the DataFrame based on a boolean mask.

Parameters:

Name	Type	Description	Default
mask	Series	A boolean Series where True indicates the rows to drop.	required

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def drop_rows(self, mask: pd.Series) -> None:
    """Drop rows from the DataFrame based on a boolean mask.

    Args:
        mask (pd.Series):
            A boolean Series where True indicates the rows to drop.

    """

    self._df = self._df[~mask]

get_match_mask

get_match_mask(
    match_with: Data | DataFrame, on_columns: list[str]
) -> Series

Find rows in this Data object that match another Data object or DataFrame.

Parameters:

Name	Type	Description	Default
match_with	Data | DataFrame	The source of rows to match against.	required
on_columns	list[str]	The columns to use for the comparison key.	required

Returns:

Type	Description
Series	pd.Series: A boolean Series where True indicates a match.

Source code in packages/pyxecm/src/pyxecm/helper/data.py

def get_match_mask(self, match_with: Data | pd.DataFrame, on_columns: list[str]) -> pd.Series:
    """Find rows in this Data object that match another Data object or DataFrame.

    Args:
        match_with (Data | pd.DataFrame):
            The source of rows to match against.
        on_columns (list[str]):
            The columns to use for the comparison key.

    Returns:
        pd.Series:
            A boolean Series where True indicates a match.

    """

    # Ternary normalization
    match_df = match_with.get_data_frame() if isinstance(match_with, Data) else match_with

    if len(on_columns) == 1:
        # Optimized path for single-column (e.g., Node 'id')
        col = on_columns[0]
        return self._df[col].isin(match_df[col])
    else:
        # Optimized path for composite keys (e.g., Edge triple)
        match_index = pd.MultiIndex.from_frame(match_df[on_columns])
        self_index = pd.MultiIndex.from_frame(self._df[on_columns])
        return self_index.isin(match_index)

xml

XML helper module.

Classes:

Name	Description
XML	Handle XML processing, e.g. to parse and update Extended ECM transport packages.

XML

Handle XML processing, e.g. to parse and update Extended ECM transport packages.

Methods:

Name	Description
remove_xml_namespace	Remove namespace from XML tag.
xml_to_dict	Parse XML string and return a dictionary without namespaces.
load_xml_file	Load an XML file into a Python list of dictionaries.
load_xml_files_from_directory	Load all XML files from a directory that matches defined file names.
load_xml_files_from_directories	Load XML files from multiple directories or zip files concurrently.
get_xml_element	Retrieve an XML Element from a string using an XPath expression.
modify_xml_element	Update the text (= content) of an XML element.
search_setting	Search a setting in an XML element and return its value.
replace_setting	Replace the value of a defined setting with a new value.
replace_in_xml_files	Replace all occurrences of the search pattern with the replace string.
extract_from_xml_files	Extract the XML subtrees using an XPath in all XML files in the directory and its subdirectories.

remove_xml_namespace classmethod

remove_xml_namespace(tag: str) -> str

Remove namespace from XML tag.

Parameters:

Name	Type	Description	Default
tag	str	The XML tag with namespace.	required

Returns:

Name	Type	Description
str	str	The tag without namespace.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def remove_xml_namespace(cls, tag: str) -> str:
    """Remove namespace from XML tag.

    Args:
        tag (str):
            The XML tag with namespace.

    Returns:
        str:
            The tag without namespace.

    """

    # In Python's ElementTree, the tag namespace
    # is put into curly braces like "{namespace}element"
    # that's why this method splits after the closing curly brace
    # and takes the last item (-1):

    return tag.split("}", 1)[-1]

xml_to_dict classmethod

xml_to_dict(
    xml_string: str,
    encode: bool = False,
    include_attributes: bool = False,
) -> dict

Parse XML string and return a dictionary without namespaces.

Parameters:

Name	Type	Description	Default
xml_string	str	The XML string to process.	required
encode	bool	True if the XML string should be encoded to UTF-8 bytes. Defaults to False.	False
include_attributes	bool	True if XML attributes should be included in the dictionary. Defaults to False.	False

Returns:

Name	Type	Description
dict	dict	The XML structure converted to a dictionary.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def xml_to_dict(cls, xml_string: str, encode: bool = False, include_attributes: bool = False) -> dict:
    """Parse XML string and return a dictionary without namespaces.

    Args:
        xml_string (str):
            The XML string to process.
        encode (bool, optional):
            True if the XML string should be encoded to UTF-8 bytes. Defaults to False.
        include_attributes (bool, optional):
            True if XML attributes should be included in the dictionary. Defaults to False.

    Returns:
        dict:
            The XML structure converted to a dictionary.

    """

    def xml_element_to_dict(element: Element) -> dict:
        """Convert XML element to dictionary.

        Args:
            element (Element):
                The XML element.

        Returns:
            dict:
                Dictionary representing the XML element

        """

        tag = cls.remove_xml_namespace(element.tag)
        children = list(element)
        node_dict = {}

        if element.attrib and include_attributes:
            node_dict.update({"@{}".format(k): v for k, v in element.attrib.items()})

        if children:
            child_dict = {}
            for child in children:
                child_tag = cls.remove_xml_namespace(child.tag)
                child_value = xml_element_to_dict(child)

                # child_dict is {child_tag: ...}, we want the value inside
                value = child_value[child_tag]

                # Handle multiple occurrences of the same tag:
                if child_tag in child_dict:
                    # If the tag already exists, ensure it becomes a list
                    if not isinstance(child_dict[child_tag], list):
                        child_dict[child_tag] = [child_dict[child_tag]]
                    child_dict[child_tag].append(value)
                else:
                    child_dict[child_tag] = value

            # Merge children into node_dict
            node_dict.update(child_dict)

        # Add text if present and meaningful
        text = element.text.strip() if element.text and element.text.strip() else None
        if text:
            if node_dict:
                node_dict["#text"] = text
            else:
                return {tag: text}

        return {tag: node_dict or text}

    if encode:
        xml_string = xml_string.encode("utf-8")
    root = etree.fromstring(xml_string)

    return xml_element_to_dict(root)

load_xml_file classmethod

load_xml_file(
    file_path: str,
    xpath: str,
    dir_name: str | None = None,
    logger: Logger = default_logger,
) -> list | None

Load an XML file into a Python list of dictionaries.

Parameters:

Name	Type	Description	Default
file_path	str	The path to XML file.	required
xpath	str	XPath to select sub-elements.	required
dir_name	str | None	Directory name to include in each dictionary, if provided. Defaults to None.	None
logger	Logger	The logging object used for all log messages.	default_logger

Returns:

Type	Description
list | None	list | None: A list of dictionaries representing the parsed XML elements, or None if an error occurs during file reading or parsing.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def load_xml_file(
    cls,
    file_path: str,
    xpath: str,
    dir_name: str | None = None,
    logger: logging.Logger = default_logger,
) -> list | None:
    """Load an XML file into a Python list of dictionaries.

    Args:
        file_path (str):
            The path to XML file.
        xpath (str):
            XPath to select sub-elements.
        dir_name (str | None, optional):
            Directory name to include in each dictionary, if provided. Defaults to None.
        logger (logging.Logger, optional):
            The logging object used for all log messages.

    Returns:
        list | None:
            A list of dictionaries representing the parsed XML elements,
            or None if an error occurs during file reading or parsing.

    """

    if not os.path.exists(file_path):
        logger.error("XML File -> %s does not exist!", file_path)
        return None

    try:
        tree = etree.parse(file_path)
        if not tree:
            logger.warning("Empty or invalid XML tree for file -> %s", file_path)
            return None

        # Extract elements using the XPath:
        elements = tree.xpath(xpath)
        if not elements:
            logger.warning(
                "No elements matched XPath -> %s in file -> '%s'",
                xpath,
                file_path,
            )
            return None

        # Convert the selected elements to dictionaries
        results = []
        tag = xpath.rsplit("/", maxsplit=1)[-1]
        for element in elements:
            element_dict = xmltodict.parse(etree.tostring(element))
            if tag in element_dict:
                element_dict = element_dict[tag]
            if dir_name:
                element_dict["directory"] = dir_name
            results.append(element_dict)

    except OSError:
        logger.error("IO Error with file -> %s", file_path)
        return None
    except etree.XMLSyntaxError:
        logger.error("XML Syntax Error in file -> %s", file_path)
        return None
    except etree.DocumentInvalid:
        logger.error("Invalid XML document -> %s", file_path)
        return None

    return results

load_xml_files_from_directory classmethod

load_xml_files_from_directory(
    path_to_root: str,
    filenames: list | None,
    xpath: str | None = None,
    logger: Logger = default_logger,
) -> list | None

Load all XML files from a directory that matches defined file names.

Then using the XPath to identify a set of elements and convert them into a Python list of dictionaries.

Parameters:

Name	Type	Description	Default
path_to_root	str	Path to the root element of the directory structure	required
filenames	list	A list of filenames. This can also be patterns like "*/en/docovw.xml". If empty all filenames ending with ".xml" are used.	required
xpath	str	The XPath to the elements we want to select. Defaults to None.	None
logger	Logger	The logging object used for all log messages.	default_logger

Returns:

Type	Description
list | None	list | None: List of dictionaries, or None if an error occurred.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def load_xml_files_from_directory(
    cls,
    path_to_root: str,
    filenames: list | None,
    xpath: str | None = None,
    logger: logging.Logger = default_logger,
) -> list | None:
    """Load all XML files from a directory that matches defined file names.

    Then using the XPath to identify a set of elements and convert them
    into a Python list of dictionaries.

    Args:
        path_to_root (str):
            Path to the root element of the
            directory structure
        filenames (list):
            A list of filenames. This can also be patterns like
            "*/en/docovw.xml". If empty all filenames ending
            with ".xml" are used.
        xpath (str, optional):
            The XPath to the elements we want to select. Defaults to None.
        logger (logging.Logger, optional):
            The logging object used for all log messages.

    Returns:
        list | None:
            List of dictionaries, or None if an error occurred.

    """

    if not filenames:
        filenames = ["*.xml"]

    try:
        # Check if the provided path is a directory or a zip file that can be extracted
        # into a directory:
        if not os.path.isdir(path_to_root) and not path_to_root.endswith(".zip"):
            logger.error(
                "The provided path -> '%s' is not a valid directory or Zip file.",
                path_to_root,
            )
            return None

        # If we have a zip file we extract it - but only if it has not been extracted before:
        if path_to_root.endswith(".zip"):
            zip_file_folder = os.path.splitext(path_to_root)[0]
            if not os.path.exists(zip_file_folder):
                logger.info(
                    "Unzipping -> '%s' into folder -> '%s'...",
                    path_to_root,
                    zip_file_folder,
                )
                try:
                    with zipfile.ZipFile(path_to_root, "r") as zfile:
                        zfile.extractall(zip_file_folder)
                except zipfile.BadZipFile:
                    logger.error(
                        "Failed to extract zip file -> '%s'",
                        path_to_root,
                    )
                    return None
                except OSError:
                    logger.error(
                        "OS error occurred while trying to extract -> '%s'",
                        path_to_root,
                    )
                    return None
            else:
                logger.info(
                    "Zip file is already extracted (path -> '%s' exists). Reusing extracted data...",
                    zip_file_folder,
                )
            path_to_root = zip_file_folder

        results = []

        # Walk through the directory
        for root, _, files in os.walk(path_to_root):
            for file_data in files:
                file_path = os.path.join(root, file_data)
                file_size = os.path.getsize(file_path)
                file_name = os.path.basename(file_path)
                dir_name = os.path.dirname(file_path)

                if any(fnmatch.fnmatch(file_path, pattern) for pattern in filenames) and file_name.endswith(".xml"):
                    logger.info(
                        "Load XML file -> '%s' of size -> %s",
                        file_path,
                        file_size,
                    )
                    elements = cls.load_xml_file(
                        file_path,
                        xpath=xpath,
                        dir_name=dir_name,
                    )
                    if elements:
                        results += elements

    except NotADirectoryError:
        logger.error(
            "The given path -> '%s' is not a directory!",
            path_to_root,
        )
        return None
    except FileNotFoundError:
        logger.error(
            "The given path -> '%s' does not exist!",
            path_to_root,
        )
        return None
    except PermissionError:
        logger.error(
            "No permission to access path -> '%s'!",
            path_to_root,
        )
        return None
    except OSError:
        logger.error("Low level OS error with file -> %s", path_to_root)
        return None

    return results

load_xml_files_from_directories classmethod

load_xml_files_from_directories(
    directories: list[str],
    filenames: list[str] | None = None,
    xpath: str | None = None,
    logger: Logger = default_logger,
) -> list[dict] | None

Load XML files from multiple directories or zip files concurrently.

Process them using XPath, and return a list of dictionaries containing the extracted elements.

This method handles multiple directories or zip files, processes XML files inside them in parallel using threads, and extracts elements that match the specified XPath. It also supports pattern matching for filenames and handles errors such as missing files or permission issues.

Parameters:

Name	Type	Description	Default
directories	list[str]	A list of directories or zip files to process. Each item can be a path to a directory or a zip file that contains XML files.	required
filenames	list[str] | None	A list of filename patterns (e.g., ["/en/docovw.xml"]) to match against the XML files. If None or empty, defaults to [".xml"].	None
xpath	str | None	An optional XPath string used to filter elements from the XML files.	None
logger	Logger	The logging object used for all log messages.	default_logger

Returns:

Type	Description
list[dict] | None	list[dict] | None: A list of dictionaries containing the extracted XML elements. Returns None if any error occurs during processing.

Raises:

Type	Description
Exception	If any error occurs during processing, such as issues with directories, files, or zip extraction.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def load_xml_files_from_directories(
    cls,
    directories: list[str],
    filenames: list[str] | None = None,
    xpath: str | None = None,
    logger: logging.Logger = default_logger,
) -> list[dict] | None:
    """Load XML files from multiple directories or zip files concurrently.

    Process them using XPath, and return a list of dictionaries containing the extracted elements.

    This method handles multiple directories or zip files, processes XML files inside them in parallel
    using threads, and extracts elements that match the specified XPath. It also supports pattern matching
    for filenames and handles errors such as missing files or permission issues.

    Args:
        directories (list[str]):
            A list of directories or zip files to process. Each item can be a path
            to a directory or a zip file that contains XML files.
        filenames (list[str] | None, optional):
            A list of filename patterns (e.g., ["*/en/docovw.xml"]) to match
            against the XML files. If None or empty, defaults to ["*.xml"].
        xpath (str | None, optional):
            An optional XPath string used to filter elements from the XML files.
        logger (logging.Logger, optional):
            The logging object used for all log messages.

    Returns:
        list[dict] | None:
            A list of dictionaries containing the extracted XML elements. Returns None
            if any error occurs during processing.

    Raises:
        Exception: If any error occurs during processing, such as issues with directories, files, or zip extraction.

    """

    # Set default for filenames if not provided
    if not filenames:
        filenames = ["*.xml"]

    results_queue = Queue()

    def process_xml_file(file_path: str) -> None:
        """Process a single XML file.

        Args:
            file_path (str):
                Path to the XML file.

        Note:
            Adds elements to the results_queue defined outside this sub-method.

        """

        try:
            file_size = os.path.getsize(file_path)
            file_name = os.path.basename(file_path)
            dir_name = os.path.dirname(file_path)

            if (
                not filenames or any(fnmatch.fnmatch(file_path, pattern) for pattern in filenames)
            ) and file_name.endswith(".xml"):
                logger.info(
                    "Load XML file -> '%s' of size -> %s",
                    file_path,
                    file_size,
                )
                elements = cls.load_xml_file(
                    file_path,
                    xpath=xpath,
                    dir_name=dir_name,
                )
                if elements:
                    results_queue.put(elements)
        except FileNotFoundError:
            logger.error("File not found -> '%s'!", file_path)
        except PermissionError:
            logger.error(
                "Permission error with file -> '%s'!",
                file_path,
            )
        except OSError:
            logger.error(
                "OS error processing file -> '%s'!",
                file_path,
            )
        except ValueError:
            logger.error(
                "Value error processing file -> '%s'!",
                file_path,
            )

    # end method process_xml_file

    def process_directory_or_zip(path_to_root: str) -> list | None:
        """Process all files in a directory or zip file.

        Args:
            path_to_root (str):
                File path to the root directory or zip file.

        """

        try:
            # Handle zip files
            if path_to_root.endswith(".zip"):
                zip_file_folder = os.path.splitext(path_to_root)[0]
                if not os.path.exists(zip_file_folder):
                    logger.info(
                        "Unzipping -> '%s' into folder -> '%s'...",
                        path_to_root,
                        zip_file_folder,
                    )
                    try:
                        with zipfile.ZipFile(path_to_root, "r") as zfile:
                            zfile.extractall(zip_file_folder)
                    except zipfile.BadZipFile:
                        logger.error(
                            "Bad zip file -> '%s'!",
                            path_to_root,
                        )
                    except zipfile.LargeZipFile:
                        logger.error(
                            "Zip file is too large to process -> '%s'!",
                            path_to_root,
                        )
                    except PermissionError:
                        logger.error(
                            "Permission error extracting zip file -> '%s'!",
                            path_to_root,
                        )
                    except OSError:
                        logger.error(
                            "OS error occurred while extracting zip file -> '%s'!",
                            path_to_root,
                        )
                    return  # Don't proceed further if zip extraction fails

                else:
                    logger.info(
                        "Zip file is already extracted (path -> '%s' exists). Reusing extracted data...",
                        zip_file_folder,
                    )
                path_to_root = zip_file_folder
            # end if path_to_root.endswith(".zip")

            # Use inner threading to process files within the directory
            with ThreadPoolExecutor(
                thread_name_prefix="ProcessXMLFile",
            ) as inner_executor:
                for root, _, files in os.walk(path_to_root):
                    for file_data in files:
                        file_path = os.path.join(root, file_data)
                        inner_executor.submit(process_xml_file, file_path)

        except FileNotFoundError:
            logger.error(
                "Directory or file not found -> '%s'!",
                path_to_root,
            )
        except PermissionError:
            logger.error(
                "Permission error with directory -> '%s'!",
                path_to_root,
            )
        except OSError:
            logger.error(
                "OS error processing path -> '%s'!",
                path_to_root,
            )
        except ValueError:
            logger.error(
                "Value error processing path -> '%s'!",
                path_to_root,
            )

    # end method process_directory_or_zip

    try:
        # Resolve wildcards in the directories list
        expanded_directories: list[str] = []
        for directory in directories:
            if "*" in directory:
                expanded_directory: list = glob.glob(directory)
                logger.info(
                    "Expanding directory -> '%s' with wildcards...",
                    directory,
                )
                expanded_directories.extend(expanded_directory)
            else:
                logger.info(
                    "Directory -> '%s' has no wildcards. Not expanding...",
                    directory,
                )
                expanded_directories.append(directory)

        # Use ThreadPoolExecutor for outer level: processing directories/zip files
        logger.info(
            "Starting %d threads for each directory or zip file...",
            len(expanded_directories),
        )
        with ThreadPoolExecutor(
            thread_name_prefix="ProcessDirOrZip",
        ) as outer_executor:
            futures = [
                outer_executor.submit(process_directory_or_zip, directory) for directory in expanded_directories
            ]

            # Wait for all futures to complete
            for future in futures:
                future.result()

        # Collect results from the queue
        logger.info("Collecting results from worker queue...")
        results = []
        while not results_queue.empty():
            results.extend(results_queue.get())
        logger.info("Done. Collected %d results.", len(results))

    except FileNotFoundError:
        logger.error(
            "Directory or file not found during execution!",
        )
        return None
    except PermissionError:
        logger.error("Permission error during execution!")
        return None
    except TimeoutError:
        logger.error(
            "Timeout occurred while waiting for threads!",
        )
        return None
    except BrokenPipeError:
        logger.error(
            "Broken pipe error occurred during thread communication!",
        )
        return None

    return results

get_xml_element classmethod

get_xml_element(xml_content: str, xpath: str) -> Element

Retrieve an XML Element from a string using an XPath expression.

Parameters:

Name	Type	Description	Default
xml_content	str	XML file as a string	required
xpath	str	XPath used to find the element.	required

Returns:

Name	Type	Description
Element	Element	The XML element (None if no element matches the XPath).

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def get_xml_element(
    cls,
    xml_content: str,
    xpath: str,
) -> Element:
    """Retrieve an XML Element from a string using an XPath expression.

    Args:
        xml_content (str):
            XML file as a string
        xpath (str):
            XPath used to find the element.

    Returns:
        Element:
            The XML element (None if no element matches the XPath).

    """

    # Parse XML content into an etree
    tree = etree.fromstring(xml_content)

    # Find the XML element specified by XPath
    element = tree.find(xpath)

    return element

modify_xml_element classmethod

modify_xml_element(
    xml_content: str,
    xpath: str,
    new_value: str,
    logger: Logger = default_logger,
) -> None

Update the text (= content) of an XML element.

Parameters:

Name	Type	Description	Default
xml_content	str	The content of an XML file.	required
xpath	str	XML Path to identify the XML element.	required
new_value	str	The new text (content).	required
logger	Logger	The logging object used for all log messages.	default_logger

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def modify_xml_element(
    cls,
    xml_content: str,
    xpath: str,
    new_value: str,
    logger: logging.Logger = default_logger,
) -> None:
    """Update the text (= content) of an XML element.

    Args:
        xml_content (str):
            The content of an XML file.
        xpath (str):
            XML Path to identify the XML element.
        new_value (str):
            The new text (content).
        logger (logging.Logger, optional):
            The logging object used for all log messages.

    """
    element = cls.get_xml_element(xml_content=xml_content, xpath=xpath)

    if element is not None:
        # Modify the XML element with the new value
        element.text = new_value
    else:
        logger.warning("XML Element -> %s not found.", xpath)

search_setting classmethod

search_setting(
    element_text: str,
    setting_key: str,
    is_simple: bool = True,
    is_escaped: bool = False,
) -> str | None

Search a setting in an XML element and return its value.

The simple case covers settings like this: "syncCandidates":true, "syncCandidates":true, In this case the setting value is a scalar like true, false, a number or none the regular expression pattern searches for a setting name in "..." (quotes) followed by a colon (:). The value is taken from what follows the colon until the next comma (,)

The more complex case is a string value that may itself have commas, so we cannot look for comma as a delimiter like in the simple case but we take the value for a string delimited by double quotes ("...")

Parameters:

Name	Type	Description	Default
element_text	str	The text to examine - typically content of an XML element.	required
setting_key	str	The name of the setting key (before the colon).	required
is_simple	bool	True if the value is scalar (not having assocs with commas). Defaults to True.	True
is_escaped	bool	True if the quotes are escaped with ". Defaults to False.	False

Returns:

Type	Description
str | None	str | None: The value of the setting or None if the setting is not found.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def search_setting(
    cls,
    element_text: str,
    setting_key: str,
    is_simple: bool = True,
    is_escaped: bool = False,
) -> str | None:
    """Search a setting in an XML element and return its value.

    The simple case covers settings like this:
    "syncCandidates":true,
    "syncCandidates":true,
    In this case the setting value is a scalar like true, false, a number or none
    the regular expression pattern searches for a setting name in "..." (quotes) followed
    by a colon (:). The value is taken from what follows the colon until the next comma (,)

    The more complex case is a string value that may itself have commas,
    so we cannot look for comma as a delimiter like in the simple case
    but we take the value for a string delimited by double quotes ("...")

    Args:
        element_text (str):
            The text to examine - typically content of an XML element.
        setting_key (str):
            The name of the setting key (before the colon).
        is_simple (bool, optional):
            True if the value is scalar (not having assocs with commas). Defaults to True.
        is_escaped (bool, optional):
            True if the quotes are escaped with ". Defaults to False.

    Returns:
        str | None:
            The value of the setting or None if the setting is not found.

    """

    if is_simple:
        pattern = r""{}":[^,]*".format(setting_key) if is_escaped else r'"{}":[^,]*'.format(setting_key)
    elif is_escaped:
        pattern = r""{}":".*"".format(setting_key)
    else:
        pattern = r'"{}":"([^"]*)"'.format(setting_key)

    match = re.search(pattern, element_text)
    if match:
        setting_line = match.group(0)
        setting_value = setting_line.split(":")[1]
        return setting_value
    else:
        return None

replace_setting classmethod

replace_setting(
    element_text: str,
    setting_key: str,
    new_value: str,
    is_simple: bool = True,
    is_escaped: bool = False,
) -> str

Replace the value of a defined setting with a new value.

The simple case covers settings like this: "syncCandidates":true, "syncCandidates":true, In this case the setting value is a scalar like true, false, a number or none the regular expression pattern searches for a setting name in "..." (quotes) followed by a colon (:). The value is taken from what follows the colon until the next comma (,)

The more complex case is a string value that may itself have commas, so we cannot look for comma as a delimiter like in the simple case but we take the value for a string delimited by double quotes ("...")

Parameters:

Name	Type	Description	Default
element_text	str	The original text of the XML element (that is to be updated).	required
setting_key	str	The name of the setting.	required
new_value	str	The new value of the setting.	required
is_simple	bool	True = value is a scalar like true, false, a number or none. Defaults to True.	True
is_escaped	bool	True if the value is surrounded with ". Defaults to False.	False

Returns:

Name	Type	Description
str	str	The updated element text.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def replace_setting(
    cls,
    element_text: str,
    setting_key: str,
    new_value: str,
    is_simple: bool = True,
    is_escaped: bool = False,
) -> str:
    """Replace the value of a defined setting with a new value.

    The simple case covers settings like this:
    "syncCandidates":true,
    "syncCandidates":true,
    In this case the setting value is a scalar like true, false, a number or none
    the regular expression pattern searches for a setting name in "..." (quotes) followed
    by a colon (:). The value is taken from what follows the colon until the next comma (,)

    The more complex case is a string value that may itself have commas,
    so we cannot look for comma as a delimiter like in the simple case
    but we take the value for a string delimited by double quotes ("...")

    Args:
        element_text (str):
            The original text of the XML element (that is to be updated).
        setting_key (str):
            The name of the setting.
        new_value (str):
            The new value of the setting.
        is_simple (bool, optional):
            True = value is a scalar like true, false, a number or none. Defaults to True.
        is_escaped (bool, optional):
            True if the value is surrounded with ". Defaults to False.

    Returns:
        str:
            The updated element text.

    """

    if is_simple:
        pattern = r""{}":[^,]*".format(setting_key) if is_escaped else r'"{}":[^,]*'.format(setting_key)
    elif is_escaped:
        pattern = r""{}":".*"".format(setting_key)
    else:
        pattern = r'"{}":"([^"]*)"'.format(setting_key)

    new_text = re.sub(pattern, new_value, element_text)

    return new_text

replace_in_xml_files classmethod

replace_in_xml_files(
    directory: str,
    search_pattern: str,
    replace_string: str,
    xpath: str = "",
    setting: str = "",
    assoc_elem: str = "",
    file_extensions: list[str] | None = None,
    logger: Logger = default_logger,
) -> bool

Replace all occurrences of the search pattern with the replace string.

This is done in all XML files in the directory and its subdirectories.

Parameters:

Name	Type	Description	Default
directory	str	Directory to traverse for XML files	required
search_pattern	str	The string to search in the XML file. This can be empty if xpath is used!	required
replace_string	str	The replacement string.	required
xpath	str	An XPath can be given to narrow down the replacement to an XML element. For now the XPath needs to be constructed in a way that it returns one or none element.	''
setting	str	Narrow down the replacement to the line that includes the setting with this name. This parameter is optional.	''
assoc_elem	str	Lookup a specific assoc element. This parameter is optional.	''
file_extensions	list[str] | None	File extensions to consider during traversal. If None, defaults to [".xml"].	None
logger	Logger	The logging object used for all log messages.	default_logger

Returns:

Name	Type	Description
bool	bool	True if a replacement happened, False otherwise

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def replace_in_xml_files(
    cls,
    directory: str,
    search_pattern: str,
    replace_string: str,
    xpath: str = "",
    setting: str = "",
    assoc_elem: str = "",
    file_extensions: list[str] | None = None,
    logger: logging.Logger = default_logger,
) -> bool:
    """Replace all occurrences of the search pattern with the replace string.

    This is done in all XML files in the directory and its subdirectories.

    Args:
        directory (str):
            Directory to traverse for XML files
        search_pattern (str):
            The string to search in the XML file.
            This can be empty if xpath is used!
        replace_string (str):
            The replacement string.
        xpath (str, optional):
            An XPath can be given to narrow down the replacement to an XML element.
            For now the XPath needs to be constructed in a way that it returns
            one or none element.
        setting (str, optional):
            Narrow down the replacement to the line that includes the setting with this name.
            This parameter is optional.
        assoc_elem (str, optional):
            Lookup a specific assoc element. This parameter is optional.
        file_extensions (list[str] | None, optional):
            File extensions to consider during traversal. If None, defaults to [".xml"].
        logger (logging.Logger, optional):
            The logging object used for all log messages.

    Returns:
        bool:
            True if a replacement happened, False otherwise

    """

    # Define the regular expression pattern to search for
    # search pattern can be empty if an xpath is used. So
    # be careful here:
    if search_pattern:
        pattern = re.compile(search_pattern)

    if not file_extensions:
        file_extensions = [".xml"]

    normalized_file_extensions: set[str] = {
        ext.lower() if ext.startswith(".") else f".{ext.lower()}" for ext in file_extensions
    }

    found = False

    # Traverse the directory and its subdirectories
    for subdir, _, files in os.walk(directory):
        for filename in files:
            file_extension = os.path.splitext(filename)[1].lower()
            # Check if the file has one of the configured extensions
            if file_extension not in normalized_file_extensions:
                continue

            # Read the contents of the file
            file_path = os.path.join(subdir, filename)

            # if xpath is given we do an intelligent replacement
            if xpath:
                xml_modified = False
                logger.debug("Replacement with xpath...")
                logger.debug(
                    "XML path -> %s, setting -> %s, assoc element -> %s",
                    xpath,
                    setting,
                    assoc_elem,
                )
                tree = etree.parse(file_path)
                if not tree:
                    logger.error(
                        "Cannot parse XML tree -> %s. Skipping...",
                        file_path,
                    )
                    continue
                root = tree.getroot()
                # find the matching XML elements using the given XPath:
                elements = root.xpath(xpath)
                if not elements:
                    logger.debug(
                        "The XML file -> %s does not have any element with the given XML path -> %s. Skipping...",
                        file_path,
                        xpath,
                    )
                    continue
                for element in elements:
                    # as XPath returns a list
                    logger.debug(
                        "Found XML element -> %s in file -> %s using xpath -> %s",
                        element.tag,
                        filename,
                        xpath,
                    )
                    # the simple case: replace the complete text of the XML element
                    if not setting and not assoc_elem:
                        logger.debug(
                            "Replace complete text of XML element -> %s from -> %s to -> %s",
                            xpath,
                            element.text,
                            replace_string,
                        )
                        element.text = replace_string
                        xml_modified = True
                    # In this case we want to set a complete value of a setting (basically replacing a whole line)
                    elif setting and not assoc_elem:
                        logger.debug(
                            "Replace single setting -> %s in XML element -> %s with new value -> %s",
                            setting,
                            xpath,
                            replace_string,
                        )
                        setting_value = cls.search_setting(
                            element.text,
                            setting,
                            is_simple=True,
                        )
                        if setting_value:
                            logger.debug(
                                "Found existing setting value -> %s",
                                setting_value,
                            )
                            # Check if the setting value needs to be surrounded by quotes.
                            # Only simplistic values like booleans or numeric values don't need quotes
                            if replace_string in ("true", "false", "none") or replace_string.isnumeric():
                                replace_setting = '"' + setting + '":' + replace_string
                            else:
                                replace_setting = '"' + setting + '":"' + replace_string + '"'
                            logger.debug(
                                "Replacement setting -> %s",
                                replace_setting,
                            )
                            element.text = cls.replace_setting(
                                element_text=element.text,
                                setting_key=setting,
                                new_value=replace_setting,
                                is_simple=True,
                            )
                            xml_modified = True
                        else:
                            logger.warning(
                                "Cannot find the value for setting -> %s. Skipping...",
                                setting,
                            )
                            continue
                    # in this case the text is just one assoc (no setting substructure)
                    elif not setting and assoc_elem:
                        logger.debug(
                            "Replace single Assoc value -> %s in XML element -> %s with -> %s",
                            assoc_elem,
                            xpath,
                            replace_string,
                        )
                        assoc_string: str = Assoc.extract_assoc_string(
                            input_string=element.text,
                        )
                        logger.debug("Assoc String -> %s", assoc_string)
                        assoc_dict = Assoc.string_to_dict(
                            assoc_string=assoc_string,
                        )
                        logger.debug("Assoc Dict -> %s", str(assoc_dict))
                        assoc_dict[assoc_elem] = replace_string  # escaped_replace_string
                        assoc_string_new: str = Assoc.dict_to_string(
                            assoc_dict=assoc_dict,
                        )
                        logger.debug(
                            "Replace assoc with -> %s",
                            assoc_string_new,
                        )
                        element.text = assoc_string_new
                        element.text = element.text.replace('"', """)
                        xml_modified = True
                    # In this case we have multiple settings with their own assocs
                    elif setting and assoc_elem:
                        logger.debug(
                            "Replace single Assoc value -> %s in setting -> %s in XML element -> %s with -> %s",
                            assoc_elem,
                            setting,
                            xpath,
                            replace_string,
                        )
                        setting_value = cls.search_setting(
                            element.text,
                            setting,
                            is_simple=False,
                        )
                        if setting_value:
                            logger.debug(
                                "Found setting value -> %s",
                                setting_value,
                            )
                            assoc_string: str = Assoc.extract_assoc_string(
                                input_string=setting_value,
                            )
                            logger.debug("Assoc String -> %s", assoc_string)
                            assoc_dict = Assoc.string_to_dict(
                                assoc_string=assoc_string,
                            )
                            logger.debug("Assoc Dict -> %s", str(assoc_dict))
                            escaped_replace_string = replace_string.replace(
                                "'",
                                "\\\\\u0027",
                            )
                            logger.debug(
                                "Escaped replacement string -> %s",
                                escaped_replace_string,
                            )
                            assoc_dict[assoc_elem] = escaped_replace_string  # escaped_replace_string
                            assoc_string_new: str = Assoc.dict_to_string(
                                assoc_dict=assoc_dict,
                            )
                            assoc_string_new = assoc_string_new.replace(
                                "'",
                                "\\u0027",
                            )
                            replace_setting = '"' + setting + '":"' + assoc_string_new + '"'
                            logger.debug(
                                "Replacement setting -> %s",
                                replace_setting,
                            )
                            # here we need to apply a "trick". It is required
                            # as regexp cannot handle the special unicode escapes \u0027
                            # we require. We first insert a placeholder "PLACEHOLDER"
                            # and let regexp find the right place for it. Then further
                            # down we use a simple search&replace to switch the PLACEHOLDER
                            # to the real value (replace() does not have the issues with unicode escapes)
                            element.text = cls.replace_setting(
                                element_text=element.text,
                                setting_key=setting,
                                new_value="PLACEHOLDER",
                                is_simple=False,
                                is_escaped=False,
                            )
                            element.text = element.text.replace(
                                "PLACEHOLDER",
                                replace_setting,
                            )
                            element.text = element.text.replace('"', """)
                            xml_modified = True
                        else:
                            logger.warning(
                                "Cannot find the value for setting -> %s. Skipping...",
                                setting,
                            )
                            continue
                if xml_modified:
                    logger.debug(
                        "XML tree has been modified. Write updated file -> %s...",
                        file_path,
                    )

                    new_contents = etree.tostring(
                        tree,
                        pretty_print=True,
                        xml_declaration=True,
                        encoding="UTF-8",
                    )
                    # we need to undo some of the stupid things tostring() did:
                    new_contents = new_contents.replace(
                        b""",
                        b""",
                    )
                    new_contents = new_contents.replace(
                        b"'",
                        b"'",
                    )
                    new_contents = new_contents.replace(b">", b">")
                    new_contents = new_contents.replace(b"<", b"<")

                    # Replace single quotes inside double quotes strings with "'" (manual escaping)
                    # This is required as we next want to replace all double quotes with single quotes
                    pattern = b'"([^"]*)"'
                    new_contents = re.sub(
                        pattern,
                        lambda m: m.group(0).replace(b"'", b"'"),
                        new_contents,
                    )

                    # Replace single quotes in XML text elements with "'"
                    # and replace double quotes in XML text elements with """
                    # This is required as we next want to replace all double quotes with single quotes
                    # to make the XML files as similar as possible with Extended ECM's format
                    pattern = b">([^<>]+?)<"
                    replacement = lambda match: match.group(0).replace(  # noqa: E731
                        b'"',
                        b"&quot;",
                    )
                    new_contents = re.sub(pattern, replacement, new_contents)
                    replacement = lambda match: match.group(0).replace(  # noqa: E731
                        b"'",
                        b"&apos;",
                    )
                    new_contents = re.sub(pattern, replacement, new_contents)

                    # Change double quotes to single quotes across the XML file - Extended ECM has it that way:
                    new_contents = new_contents.replace(b'"', b"'")

                    # Write the updated contents to the file.
                    # We DO NOT want to use tree.write() here
                    # as it would undo the manual XML tweaks we
                    # need for Extended ECM. We also need "wb"
                    # as we have bytes and not str as a data type
                    with open(file_path, "wb") as f:
                        f.write(new_contents)

                    found = True
                # end if xml_modified
            # end if xpath

            # Not using xpath - do a simple search and replace:
            else:
                logger.debug("Replacement without xpath (plain text)...")
                with open(file_path, encoding="UTF-8") as f:
                    contents = f.read()
                # Replace all occurrences of the search pattern with the replace string
                new_contents = pattern.sub(replace_string, contents)

                # Write the updated contents to the file if there were replacements
                if contents != new_contents:
                    logger.debug(
                        "Found search string -> %s in XML file -> %s. Write updated file...",
                        search_pattern,
                        file_path,
                    )
                    # Write the updated contents to the file
                    with open(file_path, "w", encoding="UTF-8") as f:
                        f.write(new_contents)
                    found = True
        # end for files
    # end for subdir

    return found

extract_from_xml_files classmethod

extract_from_xml_files(
    directory: str,
    xpath: str,
    logger: Logger = default_logger,
) -> list | None

Extract the XML subtrees using an XPath in all XML files in the directory and its subdirectories.

Parameters:

Name	Type	Description	Default
directory	str	The directory to traverse for XML files.	required
xpath	str	Used to determine XML elements to extract.	required
logger	Logger	The logging object used for all log messages.	default_logger

Returns:

Type	Description
list | None	list | None: List of dictionaries (one per extracted XML element). The list is empty if the XPath does not match any element.

Source code in packages/pyxecm/src/pyxecm/helper/xml.py

@classmethod
def extract_from_xml_files(
    cls,
    directory: str,
    xpath: str,
    logger: logging.Logger = default_logger,
) -> list | None:
    """Extract the XML subtrees using an XPath in all XML files in the directory and its subdirectories.

    Args:
        directory (str):
            The directory to traverse for XML files.
        xpath (str):
            Used to determine XML elements to extract.
        logger (logging.Logger, optional):
            The logging object used for all log messages.

    Returns:
        list | None:
            List of dictionaries (one per extracted XML element). The list is empty
            if the XPath does not match any element.

    """

    extracted_data_list = []

    # Traverse the directory and its subdirectories
    for subdir, _, files in os.walk(directory):
        for filename in files:
            # Check if the file is an XML file
            if filename.endswith(".xml"):
                # Read the contents of the file
                file_path = os.path.join(subdir, filename)

                logger.debug("Extraction with xpath -> %s...", xpath)
                tree = etree.parse(file_path)
                if not tree:
                    logger.error(
                        "Cannot parse XML file -> '%s'. Skipping...",
                        file_path,
                    )
                    continue
                root = tree.getroot()
                # find the matching XML elements using the given XPath:
                elements = root.xpath(xpath)
                if not elements:
                    logger.debug(
                        "The XML file -> %s does not have any element with the given XML path -> %s. Skipping...",
                        file_path,
                        xpath,
                    )
                    continue
                for element in elements:
                    # as XPath returns a list
                    logger.debug(
                        "Found XML element -> %s in file -> %s using xpath -> %s. Add it to result list.",
                        element.tag,
                        filename,
                        xpath,
                    )
                    extracted_content = etree.tostring(element)

                    try:
                        dict_content = xmltodict.parse(extracted_content)
                    except xmltodict.expat.ExpatError:
                        logger.error(
                            "Invalid XML syntax in file -> %s. Please check the XML file for errors.",
                            filename,
                        )
                        continue

                    extracted_data_list.append(dict_content)

    return extracted_data_list

web

Module to implement functions to execute Web Requests.

Classes:

Name	Description
HTTP	Class HTTP is used to issue HTTP request and test if hosts are reachable.

HTTP

HTTP(logger: Logger = default_logger)

Class HTTP is used to issue HTTP request and test if hosts are reachable.

Initialize the HTTP object.

Methods:

Name	Description
check_host_reachable	Check if a server / web address is reachable.
http_request	Issues an http request to a given URL.
download_file	Download a file from a URL.
human_readable_size	Return a file size in human readable form.
extract_content	Extract a string from a response of a HTTP request based on an XPath.

Source code in packages/pyxecm/src/pyxecm/helper/web.py

def __init__(
    self,
    logger: logging.Logger = default_logger,
) -> None:
    """Initialize the HTTP object."""

    if logger != default_logger:
        self.logger = logger.getChild("http")
        for logfilter in logger.filters:
            self.logger.addFilter(logfilter)

check_host_reachable

check_host_reachable(hostname: str, port: int = 80) -> bool

Check if a server / web address is reachable.

Parameters:

Name	Type	Description	Default
hostname	str	The endpoint hostname.	required
port	int	The endpoint port. Defaults to 80.	80

Returns:

Name	Type	Description
bool	bool	True is reachable, False otherwise

Source code in packages/pyxecm/src/pyxecm/helper/web.py

def check_host_reachable(self, hostname: str, port: int = 80) -> bool:
    """Check if a server / web address is reachable.

    Args:
        hostname (str):
            The endpoint hostname.
        port (int, optional):
            The endpoint port. Defaults to 80.

    Returns:
        bool:
            True is reachable, False otherwise

    """

    self.logger.debug(
        "Test if host -> '%s' is reachable on port -> %s ...",
        hostname,
        str(port),
    )
    try:
        socket.getaddrinfo(hostname, port)
    except socket.gaierror as exception:
        self.logger.warning(
            "Address-related error - cannot reach host -> %s; error -> %s",
            hostname,
            exception.strerror,
        )
        return False
    except OSError as exception:
        self.logger.warning(
            "Connection error - cannot reach host -> %s; error -> %s",
            hostname,
            exception.strerror,
        )
        return False
    else:
        self.logger.debug("Host is reachable at -> %s:%s", hostname, str(port))
        return True

http_request

http_request(
    url: str,
    method: str = "POST",
    payload: dict | None = None,
    headers: dict | None = None,
    timeout: float | None = REQUEST_TIMEOUT,
    retries: int = REQUEST_MAX_RETRIES,
    wait_time: float = REQUEST_RETRY_DELAY,
    wait_on_status: list | None = None,
    show_error: bool = True,
    stream: bool = False,
) -> dict | None

Issues an http request to a given URL.

Parameters:

Name	Type	Description	Default
url	str	The URL of the request.	required
method	str	Method of the request (POST, PUT, GET, ...). Defaults to "POST".	'POST'
payload	dict	Request payload. Defaults to None.	None
headers	dict	Request header. Defaults to None. If None then a default value defined in REQUEST_FORM_HEADERS is used.	None
timeout	float | None	The timeout in seconds. Defaults to REQUEST_TIMEOUT.	REQUEST_TIMEOUT
retries	int	The number of retries. If -1 then unlimited retries. Defaults to REQUEST_MAX_RETRIES.	REQUEST_MAX_RETRIES
wait_time	float	The number of seconds to wait after each try. Defaults to REQUEST_RETRY_DELAY.	REQUEST_RETRY_DELAY
wait_on_status	list	A list of status codes we want to wait on. If None or empty then we wait for all return codes if wait_time > 0.	None
show_error	bool	Whether to show an error or a warning message in case of an error. Defaults to True.	True
stream	bool	Enable stream for response content (e.g. for downloading large files). Defaults to False.	False

Returns:

Type	Description
dict | None	dict | None: Response of call

Source code in packages/pyxecm/src/pyxecm/helper/web.py

def http_request(
    self,
    url: str,
    method: str = "POST",
    payload: dict | None = None,
    headers: dict | None = None,
    timeout: float | None = REQUEST_TIMEOUT,
    retries: int = REQUEST_MAX_RETRIES,
    wait_time: float = REQUEST_RETRY_DELAY,
    wait_on_status: list | None = None,
    show_error: bool = True,
    stream: bool = False,
) -> dict | None:
    """Issues an http request to a given URL.

    Args:
        url (str):
            The URL of the request.
        method (str, optional):
            Method of the request (POST, PUT, GET, ...). Defaults to "POST".
        payload (dict, optional):
            Request payload. Defaults to None.
        headers (dict, optional):
            Request header. Defaults to None. If None then a default
            value defined in REQUEST_FORM_HEADERS is used.
        timeout (float | None, optional):
            The timeout in seconds. Defaults to REQUEST_TIMEOUT.
        retries (int, optional):
            The number of retries. If -1 then unlimited retries.
            Defaults to REQUEST_MAX_RETRIES.
        wait_time (float, optional):
            The number of seconds to wait after each try.
            Defaults to REQUEST_RETRY_DELAY.
        wait_on_status (list, optional):
            A list of status codes we want to wait on.
            If None or empty then we wait for all return codes if
            wait_time > 0.
        show_error (bool, optional):
            Whether to show an error or a warning message in case of an error.
            Defaults to True.
        stream (bool, optional):
            Enable stream for response content (e.g. for downloading large files).
            Defaults to False.

    Returns:
        dict | None:
            Response of call

    """

    if not headers:
        headers = REQUEST_FORM_HEADERS

    message = "Make HTTP request to URL -> '{}' using -> {} method".format(
        url,
        method,
    )
    if payload:
        message += " with payload -> {}".format(payload)
    if retries:
        message += " (max number of retries -> {}, wait time between retries -> {})".format(
            retries,
            wait_time,
        )
        try:
            retries = int(retries)
        except ValueError:
            self.logger.warning(
                "HTTP request -> retries is not a valid integer value: %s, defaulting to 0 retries ",
                retries,
            )
            retries = 0

    self.logger.debug(message)

    try_counter = 1

    while True:
        try:
            response = requests.request(
                method=method,
                url=url,
                data=payload,
                headers=headers,
                timeout=timeout,
                stream=stream,
            )
        except requests.RequestException as exc:
            response = None
            self.logger.warning(
                "HTTP request -> %s to url -> %s failed (try %s); error -> %s",
                method,
                url,
                try_counter,
                exc,
            )

        if response is not None:
            # Do we have a successful result?
            if response.ok:
                self.logger.debug(
                    "HTTP request -> %s to url -> %s succeeded with status -> %s!",
                    method,
                    url,
                    response.status_code,
                )

                if wait_on_status and response.status_code in wait_on_status:
                    self.logger.debug(
                        "%s is in wait_on_status list -> %s",
                        response.status_code,
                        wait_on_status,
                    )
                else:
                    return response

            else:
                message = "HTTP request -> {} to url -> {} failed; status -> {}; error -> {}".format(
                    method,
                    url,
                    response.status_code,
                    (
                        response.text
                        if response.headers.get("content-type") == "application/json"
                        else "see debug log"
                    ),
                )
                if show_error and retries == 0:
                    self.logger.error(message)
                else:
                    self.logger.warning(message)
        # end if response is not None

        # Check if another retry is allowed, if not return None
        if retries == 0:
            return None

        if wait_time > 0.0:
            self.logger.warning(
                "Sleeping %s seconds and then trying once more...",
                str(wait_time * try_counter),
            )
            time.sleep(wait_time * try_counter)

        retries -= 1
        try_counter += 1

download_file

download_file(
    url: str,
    filename: str,
    timeout: float = REQUEST_TIMEOUT,
    retries: int | None = REQUEST_MAX_RETRIES,
    wait_time: float = REQUEST_RETRY_DELAY,
    wait_on_status: list | None = None,
    chunk_size: int = 8192,
    show_error: bool = True,
) -> bool

Download a file from a URL.

Parameters:

Name	Type	Description	Default
url	str	The URL to open / load.	required
filename	str	The filename to save the content.	required
timeout	float	The timeout in seconds. Defaults to REQUEST_TIMEOUT.	REQUEST_TIMEOUT
retries	int | None	The number of retries. If -1 then unlimited retries. Defaults to REQUEST_MAX_RETRIES.	REQUEST_MAX_RETRIES
wait_time	float	The number of seconds to wait after each try. Defaults to REQUEST_RETRY_DELAY.	REQUEST_RETRY_DELAY
wait_on_status	list	The list of status codes we want to wait on. If None or empty then we wait for all return codes if wait_time > 0.	None
chunk_size	int	Chunk size in bytes for reading file content. Default is 8192.	8192
show_error	bool	Whether or not an error should be logged if the download fails. Default is True.	True

Returns:

Name	Type	Description
bool	bool	True if successful, False otherwise.

Source code in packages/pyxecm/src/pyxecm/helper/web.py

def download_file(
    self,
    url: str,
    filename: str,
    timeout: float = REQUEST_TIMEOUT,
    retries: int | None = REQUEST_MAX_RETRIES,
    wait_time: float = REQUEST_RETRY_DELAY,
    wait_on_status: list | None = None,
    chunk_size: int = 8192,
    show_error: bool = True,
) -> bool:
    """Download a file from a URL.

    Args:
        url (str):
            The URL to open / load.
        filename (str):
            The filename to save the content.
        timeout (float, optional):
            The timeout in seconds. Defaults to REQUEST_TIMEOUT.
        retries (int | None, optional):
            The number of retries. If -1 then unlimited retries.
            Defaults to REQUEST_MAX_RETRIES.
        wait_time (float, optional):
            The number of seconds to wait after each try.
            Defaults to REQUEST_RETRY_DELAY.
        wait_on_status (list, optional):
            The list of status codes we want to wait on.
            If None or empty then we wait for all return codes if
            wait_time > 0.
        chunk_size (int, optional):
            Chunk size in bytes for reading file content. Default is 8192.
        show_error (bool, optional):
            Whether or not an error should be logged if the download fails.
            Default is True.

    Returns:
        bool:
            True if successful, False otherwise.

    """

    # Validate the URL:
    parsed_url = urlparse(url)
    if not parsed_url.scheme or not parsed_url.netloc:
        self.logger.error("Invalid URL -> '%s' to download a file!", url)
        return False

    response = self.http_request(
        url=url,
        method="GET",
        retries=retries,
        timeout=timeout,
        wait_time=wait_time,
        wait_on_status=wait_on_status,
        show_error=show_error,
        stream=True,  # for downloads we want streaming
    )

    if not response or not response.ok:
        self.logger.error(
            "Failed to request download file -> '%s' from site -> %s%s",
            filename,
            url,
            "; error -> {}".format(response.text) if response else "",
        )
        return False

    try:
        directory = os.path.dirname(filename)
        if not os.path.exists(directory):
            self.logger.info(
                "Download directory -> '%s' does not exist, creating it.",
                directory,
            )
            os.makedirs(directory)
        with open(filename, "wb") as download_file:
            download_file.writelines(response.iter_content(chunk_size=chunk_size))
        self.logger.debug(
            "File downloaded successfully as -> '%s' (size -> %s).",
            filename,
            self.human_readable_size(os.path.getsize(filename)),
        )
    except (OSError, requests.exceptions.RequestException):
        self.logger.error(
            "Cannot write content to file -> '%s' in directory -> '%s'!",
            filename,
            directory,
        )
        return False
    else:
        return True

human_readable_size

human_readable_size(size_in_bytes: int) -> str

Return a file size in human readable form.

Parameters:

Name	Type	Description	Default
size_in_bytes	int	The file size in bytes.	required

Returns:

Name	Type	Description
str	str	The formatted size using units.

Source code in packages/pyxecm/src/pyxecm/helper/web.py

def human_readable_size(self, size_in_bytes: int) -> str:
    """Return a file size in human readable form.

    Args:
        size_in_bytes (int): The file size in bytes.

    Returns:
        str:
            The formatted size using units.

    """

    for unit in ["B", "KB", "MB", "GB", "TB"]:
        if size_in_bytes < 1024:
            return "{:.2f} {}".format(size_in_bytes, unit)
        size_in_bytes /= 1024

    # We should never get here but linter wants it:
    return "{:.2f}".format(size_in_bytes)

extract_content

extract_content(url: str, xpath: str) -> str | None

Extract a string from a response of a HTTP request based on an XPath.

Parameters:

Name	Type	Description	Default
url	str	The URL to open / load.	required
xpath	str	The XPath expression to apply to the result.	required

Returns:

Type	Description
str | None	str | None: Extracted string or None in case of an error.

Source code in packages/pyxecm/src/pyxecm/helper/web.py

def extract_content(self, url: str, xpath: str) -> str | None:
    """Extract a string from a response of a HTTP request based on an XPath.

    Args:
        url (str):
            The URL to open / load.
        xpath (str):
            The XPath expression to apply to the result.

    Returns:
        str | None:
            Extracted string or None in case of an error.

    """

    # Send a GET request to the URL:
    response = self.http_request(
        url=url,
        method="GET",
    )

    # Check if request was successful
    if response and response.ok:
        # Parse the HTML content
        tree = html.fromstring(response.content)

        # Extract content using XPath
        elements = tree.xpath(xpath)

        # Get text content of all elements and join them
        content = "\n".join([elem.text_content().strip() for elem in elements])

        # Return the extracted content:
        return content

    # If request was not successful, print error message:
    self.logger.error(
        "Cannot extract content from URL -> '%s'; error code -> %s",
        url,
        response.status_code,
    )

    return None

logadapter

Custom adapter to prefix all messages with a custom prefix.

Classes:

Name	Description
PrefixLogAdapter	Prefix all messages with a custom prefix.

PrefixLogAdapter

Bases: LoggerAdapter

Prefix all messages with a custom prefix.

Methods:

Name	Description
process	Prefix the log message with the value of extra["prefix"] in square brackets.

process

process(msg: str, kwargs: dict) -> tuple[str, dict]

Prefix the log message with the value of extra["prefix"] in square brackets.

Parameters:

Name	Type	Description	Default
msg	str	The log message.	required
kwargs	dict	The keyword arguments of the logging call. Returned unchanged.	required

Returns:

Type	Description
tuple[str, dict]	tuple[str, dict]: The prefixed message and the unchanged keyword arguments.

Source code in packages/pyxecm/src/pyxecm/helper/logadapter.py

def process(self, msg: str, kwargs: dict) -> tuple[str, dict]:
    """Prefix the log message with the value of `extra["prefix"]` in square brackets.

    Args:
        msg (str):
            The log message.
        kwargs (dict):
            The keyword arguments of the logging call. Returned unchanged.

    Returns:
        tuple[str, dict]:
            The prefixed message and the unchanged keyword arguments.

    """

    return "[{}] {}".format(self.extra["prefix"], msg), kwargs