sensospot_parser/sensospot_data/parameters.py

""" Sensospot Data Parser

Parsing the numerical output from Sensovations Sensospot image analysis.
"""

import pathlib
from typing import Any, Dict, Union, Optional
from xml.etree.ElementTree import Element as ElementType

import numpy
import pandas
from defusedxml import ElementTree

from . import columns

PathLike = Union[str, pathlib.Path]


def _search_params_file(folder: PathLike) -> Optional[PathLike]:
    """searches for a exposure settings file in a folder"""
    folder_path = pathlib.Path(folder)
    params_folder = folder_path / "Parameters"
    if not params_folder.is_dir():
        return None
    param_files = list(params_folder.glob("**/*.svexp"))
    if len(param_files) == 1:
        return param_files[0]
    else:
        return None


def _get_channel_data(channel_node: ElementType) -> Dict[str, Any]:
    # child.tag == "ChannelConfig1"
    exposure_id = int(channel_node.tag[-1])
    # channel_description == "[Cy3|Cy5] Green"
    description = channel_node.attrib["Description"]
    exposure_channel = description.rsplit(" ", 1)[-1]
    # floats can be used for exposure times, not only ints
    exposure_time = float(channel_node.attrib["ExposureTimeMs"])
    return {
        columns.EXPOSURE_ID: exposure_id,
        columns.PARAMETERS_CHANNEL: exposure_channel.lower(),
        columns.PARAMETERS_TIME: exposure_time,
    }


def _parse_measurement_params(params_file: PathLike) -> pandas.DataFrame:
    """parses the cannel informations from a settings file"""
    file_path = pathlib.Path(params_file)
    with file_path.open("r") as file_handle:
        tree = ElementTree.parse(file_handle)
        data = [_get_channel_data(child) for child in tree.find("Channels")]
    return pandas.DataFrame(data)


def get_measurement_params(folder: PathLike) -> Optional[pandas.DataFrame]:
    """returns measurement parameters"""
    params_file = _search_params_file(folder)
    if params_file is not None:
        return _parse_measurement_params(params_file)
    return None


def add_measurement_parameters(
    data_frame: pandas.DataFrame, folder: PathLike
) -> pandas.DataFrame:
    """adds measurement params to the data frame, if they could be parsed"""
    params = get_measurement_params(folder)
    if params is not None:
        params_exposures = params[columns.EXPOSURE_ID].unique()
        data_exposures = data_frame[columns.EXPOSURE_ID].unique()
        if set(data_exposures) == set(params_exposures):
            return data_frame.merge(params, how="left", on=columns.EXPOSURE_ID)

    # only executing if the parameters were not merged to the data frame
    data_frame[columns.PARAMETERS_CHANNEL] = numpy.nan
    data_frame[columns.PARAMETERS_TIME] = numpy.nan
    return data_frame