|
|
@ -16,8 +16,12 @@ from . import columns |
|
|
|
PathLike = Union[str, pathlib.Path] |
|
|
|
PathLike = Union[str, pathlib.Path] |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _search_params_file(folder: PathLike) -> Optional[PathLike]: |
|
|
|
def _search_params_file(folder: PathLike) -> Optional[pathlib.Path]: |
|
|
|
"""searches for a exposure settings file in a folder""" |
|
|
|
"""searches for a exposure settings file in a folder |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
folder: directory to search |
|
|
|
|
|
|
|
returns: the path to the settings file or None |
|
|
|
|
|
|
|
""" |
|
|
|
folder_path = pathlib.Path(folder) |
|
|
|
folder_path = pathlib.Path(folder) |
|
|
|
params_folder = folder_path / "Parameters" |
|
|
|
params_folder = folder_path / "Parameters" |
|
|
|
if not params_folder.is_dir(): |
|
|
|
if not params_folder.is_dir(): |
|
|
@ -30,6 +34,11 @@ def _search_params_file(folder: PathLike) -> Optional[PathLike]: |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _get_channel_data(channel_node: ElementType) -> Dict[str, Any]: |
|
|
|
def _get_channel_data(channel_node: ElementType) -> Dict[str, Any]: |
|
|
|
|
|
|
|
"""parses the information from an xml node of the channel settings |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
channel_node: the xml node of the channel settings |
|
|
|
|
|
|
|
returns: dict with the information |
|
|
|
|
|
|
|
""" |
|
|
|
# child.tag == "ChannelConfig1" |
|
|
|
# child.tag == "ChannelConfig1" |
|
|
|
exposure_id = int(channel_node.tag[-1]) |
|
|
|
exposure_id = int(channel_node.tag[-1]) |
|
|
|
# channel_description == "[Cy3|Cy5] Green" |
|
|
|
# channel_description == "[Cy3|Cy5] Green" |
|
|
@ -45,7 +54,11 @@ def _get_channel_data(channel_node: ElementType) -> Dict[str, Any]: |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _parse_measurement_params(params_file: PathLike) -> pandas.DataFrame: |
|
|
|
def _parse_measurement_params(params_file: PathLike) -> pandas.DataFrame: |
|
|
|
"""parses the cannel informations from a settings file""" |
|
|
|
"""parses the cannel informations from a settings file |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
params_file: path to the settings file |
|
|
|
|
|
|
|
returns: pandas DataFrame with the parsed information |
|
|
|
|
|
|
|
""" |
|
|
|
file_path = pathlib.Path(params_file) |
|
|
|
file_path = pathlib.Path(params_file) |
|
|
|
with file_path.open("r") as file_handle: |
|
|
|
with file_path.open("r") as file_handle: |
|
|
|
tree = ElementTree.parse(file_handle) |
|
|
|
tree = ElementTree.parse(file_handle) |
|
|
@ -54,7 +67,11 @@ def _parse_measurement_params(params_file: PathLike) -> pandas.DataFrame: |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def get_measurement_params(folder: PathLike) -> Optional[pandas.DataFrame]: |
|
|
|
def get_measurement_params(folder: PathLike) -> Optional[pandas.DataFrame]: |
|
|
|
"""returns measurement parameters""" |
|
|
|
"""searches the settings file and returns the parameters |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
folder: path to the folder with the measurement data |
|
|
|
|
|
|
|
returns: pandas DataFrame with the parsed parameters or None |
|
|
|
|
|
|
|
""" |
|
|
|
params_file = _search_params_file(folder) |
|
|
|
params_file = _search_params_file(folder) |
|
|
|
if params_file is not None: |
|
|
|
if params_file is not None: |
|
|
|
return _parse_measurement_params(params_file) |
|
|
|
return _parse_measurement_params(params_file) |
|
|
@ -62,17 +79,29 @@ def get_measurement_params(folder: PathLike) -> Optional[pandas.DataFrame]: |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def add_measurement_parameters( |
|
|
|
def add_measurement_parameters( |
|
|
|
data_frame: pandas.DataFrame, folder: PathLike |
|
|
|
measurement: pandas.DataFrame, folder: PathLike |
|
|
|
) -> pandas.DataFrame: |
|
|
|
) -> pandas.DataFrame: |
|
|
|
"""adds measurement params to the data frame, if they could be parsed""" |
|
|
|
"""adds measurement params to the data frame, if they could be parsed |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
The returned DataFrame will contain two more columns for parsed time and |
|
|
|
|
|
|
|
channel from the parameters file. |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
If the parameters could not be found, parsed or do not match up with the |
|
|
|
|
|
|
|
measurement data, the additional collumns will contain NaN. |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
measurement: the parsed measurement data |
|
|
|
|
|
|
|
returns: the measurement data with parameters added |
|
|
|
|
|
|
|
""" |
|
|
|
params = get_measurement_params(folder) |
|
|
|
params = get_measurement_params(folder) |
|
|
|
if params is not None: |
|
|
|
if params is not None: |
|
|
|
params_exposures = params[columns.EXPOSURE_ID].unique() |
|
|
|
params_exposures = params[columns.EXPOSURE_ID].unique() |
|
|
|
data_exposures = data_frame[columns.EXPOSURE_ID].unique() |
|
|
|
data_exposures = measurement[columns.EXPOSURE_ID].unique() |
|
|
|
if set(data_exposures) == set(params_exposures): |
|
|
|
if set(data_exposures) == set(params_exposures): |
|
|
|
return data_frame.merge(params, how="left", on=columns.EXPOSURE_ID) |
|
|
|
return measurement.merge( |
|
|
|
|
|
|
|
params, how="left", on=columns.EXPOSURE_ID |
|
|
|
|
|
|
|
) |
|
|
|
|
|
|
|
|
|
|
|
# only executing if the parameters were not merged to the data frame |
|
|
|
# only executing if the parameters were not merged to the data frame |
|
|
|
data_frame[columns.PARAMETERS_CHANNEL] = numpy.nan |
|
|
|
measurement[columns.PARAMETERS_CHANNEL] = numpy.nan |
|
|
|
data_frame[columns.PARAMETERS_TIME] = numpy.nan |
|
|
|
measurement[columns.PARAMETERS_TIME] = numpy.nan |
|
|
|
return data_frame |
|
|
|
return measurement |
|
|
|