bat-studio/modules/parsers/nbus_stream_parser.py

import pandas as pd
import numpy as np


class DataParser:
    def __init__(self, dataframe: pd.DataFrame):
        self._df = dataframe

    def remap_columns(self, new_columns):
        self._df.rename(columns=new_columns, inplace=True)

    def remap_nans(self, new_nans):
        # Loop through the column names and their target float values
        for col, target_value in new_nans.items():

            # Check if the column exists in your DataFrame to prevent KeyErrors
            if col in self._df.columns:

                # 1. Convert the column to numeric type (strings/text will be coerced to NaN)
                self._df[col] = pd.to_numeric(self._df[col], errors='coerce')

                # 2. Create a safe float mask for just this column
                equal_to = np.isclose(
                    self._df[col],
                    target_value,
                    atol=1e-8
                )

                # 3. Apply the mask and replace matching values with NaN in-place
                self._df.loc[equal_to, col] = np.nan

    def validate_and_filter_series_by_zeros(self, ser_col, pha_col, expected_phases):
        """
        Validates measurement series separated by zeros. If a series block between
        zeros does not contain the expected number of unique phases, the entire block is removed.

        Parameters:
        - ser_col (str): Column name for the series ID.
        - pha_col (str): Column name for the phase ID.
        - expected_phases (int): Required number of unique phases per series block (default: 6).
        """

        # 1. Identify rows that are zeros (separator rows)
        is_zero = (self._df[ser_col] == 0) & (self._df[pha_col] == 0)

        # 2. Create a unique group ID for each block of rows between zeros
        block_id = is_zero.cumsum()

        valid_indices = []

        # 3. Iterate through each block between zeros
        ser_no = 1

        for b_id, group in self._df.groupby(block_id):

            # Skip pure zero separator blocks (we can keep or drop them, usually they are separators)
            if is_zero.loc[group.index].all():
                continue

            # Check how many unique phases are in this specific block
            unique_phases = group[pha_col].nunique() - 1

            # If it matches the expected number of phases, keep its indices
            if unique_phases == expected_phases:
                self._df.loc[group.index, ser_col] = ser_no
                ser_no += 1
                valid_indices.extend(group.index.tolist())

        # 4. Keep only rows belonging to valid blocks
        invalid_indices = self._df.index.difference(valid_indices)
        self._df.drop(index=invalid_indices, inplace=True)

    def remove_specific_phases(self, pha_col, phases_to_remove):
        self._df.drop(
            index=self._df.index[
                self._df[pha_col].isin(phases_to_remove)
            ],
            inplace=True
        )

    def remap_phases(self, pha_col, phase_mapping):
        # Replace phase values according to the dictionary mapping
        if phase_mapping is not None:
            self._df[pha_col] = self._df[pha_col].replace(phase_mapping)

    def drop_rows_completely_null(self, columns):
        """
        Drops rows where ALL columns (except series and phase) are NaN simultaneously.
        """

        cols_to_check = [
            col for col in self._df.columns if col in columns
        ]

        self._df.dropna(
            subset=cols_to_check,
            how='all',
            inplace=True
        )

    def clean_precision_noise(self, precision=5):
        """
        Finds all float columns, rounds off float precision artifacts (e.g., 3.0000002 -> 3.0),
        and safely converts them using the 'g' specifier format so trailing zero decimals
        are completely dropped upon viewing or CSV export, while leaving real decimals intact.
        """

        # 1. Locate all float columns dynamically to avoid breaking string/text columns
        float_cols = self._df.select_dtypes(
            include=['float', 'float64']
        ).columns

        # 2. Round the target columns to snap the noisy precision bits to their actual values
        self._df[float_cols] = self._df[float_cols].round(precision)

        # 3. Format the cells to trim the trailing '.0' cleanly using g-formatting
        # (This is vectorized across selected fields and leaves NaNs unharmed)
        for col in float_cols:
            self._df[col] = self._df[col].map(
                lambda x: f"{x:g}" if pd.notnull(x) else x
            )

    def get_dataframe(self):
        return self._df