external_file_data.py

"""
ExternalFileData class to read data from an external file using pandas.
"""

import logging

import pandas as pd


class ExternalFileData:
    """
    Class to read data from an external file using pandas.
    """

    def __init__(self, file_path: str, parameters: dict):
        """
        Initialize the ExternalFileData class.
        :param file_path: Path to the external file.
        :param parameters: Parameters for reading the file (e.g., delimiter, header). Check pd.read_csv for details.
        """
        self.__file_path: str = file_path
        self.__parameters: dict = parameters
        self.__df: pd.DataFrame | None = None
        self.__log: logging.Logger = logging.getLogger(__name__)

        self.__units: list[str] | None = None
        self.__names: list[str] | None = None
        self.__descriptions: list[str] | None = None

    def close(self):
        """
        Close the file and release resources.
        """
        if self.__df is not None:
            self.__log.info("Closing file: %s", self.__file_path)
            self.__df = None

    def not_my_file(self) -> bool:
        """
        Check if the file should be read with this plugin.
        :return: True if the file should not be read with this plugin, False otherwise.
        """
        if self.data() is None:
            return True

        return ExternalFileData._not_my_file(self.__names, self.__descriptions, self.__units)

    @staticmethod
    def _not_my_file(names, descriptions, units) -> bool:
        if names is None or descriptions is None or units is None:
            return True

        nr_of_channels = len(names)
        nr_of_descriptions = len(descriptions)
        nr_of_units = len(units)
        if nr_of_channels < 2 or nr_of_channels != nr_of_descriptions or nr_of_channels != nr_of_units:
            return True

        # first column in dydaqlog CSV file is "Sequence Number"
        if nr_of_descriptions == 0 or "sequence number" != descriptions[0].lower():
            return True

        return False

    def data(self) -> pd.DataFrame:
        """
        Read the data from the file and return it as a pandas DataFrame.
        :return: DataFrame containing the data from the file.
        """
        if self.__df is None:
            self.__log.info("Reading file: %s", self.__file_path)

            # The first three rows are containing the metadata
            with open(self.__file_path, "r", encoding="utf-8-sig") as file:
                self.__names = [item.strip().strip('"') for item in file.readline().split(";")]
                self.__descriptions = [item.strip().strip('"') for item in file.readline().split(";")]
                self.__units = [item.strip().strip('"') for item in file.readline().split(";")]

            if not ExternalFileData._not_my_file(self.__names, self.__descriptions, self.__units):
                # Read the data from the file - skip the header (already read)
                df = pd.read_csv(self.__file_path, sep=";", decimal=",", header=None, skiprows=3)
                df[2] = pd.to_datetime(df[2] + "T" + df[3] + "." + df[4].astype(str))
                self.__df = df

        return self.__df

    def column_names(self) -> list[str] | None:
        """
        Overwrite the column names of the dataframe.
        """
        if self.__names is None:
            self.data()

        return self.__names

    def column_units(self) -> list[str] | None:
        """
        Return column units of the dataframe.
        """
        if self.__units is None:
            self.data()

        return self.__units

    def column_descriptions(self) -> list[str] | None:
        """
        Return column descriptions of the dataframe.
        """
        if self.__descriptions is None:
            self.data()

        return self.__descriptions