Source code for bhf_smarthealth.log.metadata_logs

"""This module contains functions to generate metadata logs."""

import logging
import pandas as pd

from datetime import date
from pathlib import Path
from collections import Counter

from bhf_smarthealth.models import DataStreamNames
from bhf_smarthealth.utils import data_io
from bhf_smarthealth.config import DEFAULT_VALID_DATA_THRESHOLD
from bhf_smarthealth.exceptions import InsufficientValidDataError

logger = logging.getLogger(__name__)


# TODO: Write unit tests for this module




class DataQualityCounter(Counter[str]):
    """Custom collections.Counter instance to accept only allowed data quality categories as keys"""

    categories = ["missing", "empty", "valid", "processing_error"]

    def __setitem__(self, key: str, value: int) -> None:
        if key.lower() not in self.categories:
            raise KeyError(
                f"Invalid data quality category: {key}. Category should be one of: {', '.join(self.categories)}"
            )
        return super().__setitem__(key, value)





def evaluate_daily_processing_success(
    counter: DataQualityCounter,
    validity_threshold: float = DEFAULT_VALID_DATA_THRESHOLD,
):
    """Evaluate the data quality in a daily data batch. Logs information about the number of successfully-processed files if the specified threshold is satisfied, else an error is raised.

    Parameters
    ----------
    counter : DataQualityCounter
        custom collections.Counter instance that accepts only allowed data quality category keys
    validity_threshold : float, optional
        Ratio of invalid : total files allowed to pass through the pipeline, by default DEFAULT_VALID_DATA_THRESHOLD

    Raises
    ------
    InsufficientValidDataError
        Raised if an insufficient number of daily data files were successfully processed
    """
    total_files = counter.total()
    total_invalid = total_files - counter["valid"]

    if total_files == counter["valid"]:
        logger.info(f"Successfully processed all {total_files} files")

    elif counter["valid"] / total_files < validity_threshold:
        raise InsufficientValidDataError(
            f"The number of invalid files ({total_invalid}) out of {total_files} total files, corresponding to a data validity ratio of {round(counter['valid'] / total_files, 2)}, does not meet the specified threshold for data validity ({validity_threshold}).\nThe breakdown of invalid files is: {counter}"
        )

    else:
        logger.warning(
            f"Only {counter['valid']} files were successfully processed out of {total_files}, corresponding to a data validity ratio of {round(counter['valid'] / total_files, 2)}"
        )





def generate_daily_IDs_receipt(
    all_IDs: list[str],
    valid_IDs: list[str],
    data_stream: DataStreamNames,
    date_str: str,
    out_path: Path,
) -> None:
    """Generate a .parquet receipt containing all of the sheffield_ids recieved on a single given date the specified data_type, and whether the data was valid on that date.

    Parameters
    ----------
    all_IDs : list[str]
        Full list of strings of IDs for which data has been recieved in a given day
    valid_IDs : list[str]
        List of strings of IDs for which valid data has been successfully processed in a given day
    data_stream : DataStreamNames
        Data stream name
    date_str : str
        Date string passed through snakemake 'date' wildcard in form %Y%m%d
    out_path : Path
       Path to save .parquet file to
    """
    date_obj = date.fromisoformat(date_str)

    df = pd.DataFrame(
        {"sheffield_id": all_IDs, "data_stream": data_stream, "event_date": date_obj}
    )

    df["is_valid"] = df["sheffield_id"].apply(
        lambda x: True if x in valid_IDs else False
    )

    data_io.save_dataframe_ouput(df=df, file_out_path=out_path)