feat: Luminex xPONENT - add errors for nan and missing values in statistics tables (#1036)

Author: slopez-b

src/allotropy/parsers/luminex_xponent/luminex_xponent_reader.py

@@ -91,9 +91,15 @@ def _get_results(reader: CsvReader) -> dict[str, pd.DataFrame]:
             result_title = match.groups()[0]
             table_data = assert_not_none(
                 reader.pop_csv_block_as_df(
-                    empty_pat=constants.LUMINEX_EMPTY_PATTERN, header=0, index_col=0
+                    empty_pat=constants.LUMINEX_EMPTY_PATTERN,
+                    header=0,
+                    index_col=0,
+                    na_values=["NaN", "nan", "-NaN", "-nan", "None", "null", "NULL"],
+                    keep_default_na=False,  # Prevent pandas from interpreting empty strings as NaN.
                 )
-            ).dropna(how="all", axis="columns")
+            )
+            # drop empty columns at the end of the table
+            table_data = table_data.loc[:, ~table_data.columns.str.contains("^Unnamed")]
             results[result_title] = table_data
             reader.drop_empty(constants.LUMINEX_EMPTY_PATTERN)
 

src/allotropy/parsers/luminex_xponent/luminex_xponent_structure.py

@@ -4,6 +4,7 @@
 import itertools
 from pathlib import Path
 import re
+from typing import Any
 
 import pandas as pd
 
@@ -39,7 +40,12 @@
 )
 from allotropy.parsers.utils.pandas import map_rows, SeriesData
 from allotropy.parsers.utils.uuids import random_uuid_str
-from allotropy.parsers.utils.values import assert_not_none, try_float
+from allotropy.parsers.utils.values import (
+    assert_not_none,
+    try_float,
+    try_non_nan_float_or_negative_zero,
+    try_non_nan_float_or_none,
+)
 
 
 @dataclass(frozen=True)
@@ -163,6 +169,7 @@ def create(
         location = str(count_data.series.name)
         dilution_factor_data = results_data["Dilution Factor"]
         errors_data = results_data.get("Warnings/Errors")
+        measurement_identifier = random_uuid_str()
 
         if location not in dilution_factor_data.index:
             msg = f"Could not find 'Dilution Factor' data for: '{location}'."
@@ -177,31 +184,49 @@ def create(
         )
         errors: list[Error] = []
         data_errors = cls._get_errors(errors_data, well_location) or []
-        for error in data_errors:
-            errors.append(Error(error=error))
+        for data_error in data_errors:
+            errors.append(Error(error=data_error))
 
         if dilution_factor_setting == NEGATIVE_ZERO:
             errors.append(
                 Error(error="Not reported in file", feature="dilution factor setting")
             )
 
+        def get_statistics_error(
+            raw_value: Any, analyte: str, role: str
+        ) -> Error | None:
+            """Check if the raw value is empty or NaN and return an Error if so."""
+            error = None
+            if raw_value == "":
+                error = "Not reported"
+            elif try_non_nan_float_or_none(raw_value) is None:
+                error = "NaN"
+            return Error(error=error, feature=f"{analyte} <{role}>") if error else None
+
         def get_statistic_dimensions(analyte: str) -> list[StatisticDimension]:
-            return [
-                StatisticDimension(
-                    value=try_float(statistic_table.at[location, analyte], analyte),
-                    unit=statistic_conf.unit,
-                    statistic_datum_role=statistic_conf.role,
+            statistic_dimensions = []
+            for section, statistic_conf in STATISTIC_SECTIONS_CONF.items():
+                if (statistic_table := results_data.get(section)) is None:
+                    continue
+                raw_value = statistic_table.at[location, analyte]
+                role = statistic_conf.role
+                if error := get_statistics_error(raw_value, analyte, role.value):
+                    errors.append(error)
+                statistic_dimensions.append(
+                    StatisticDimension(
+                        value=try_non_nan_float_or_negative_zero(raw_value),
+                        unit=statistic_conf.unit,
+                        statistic_datum_role=role,
+                    )
                 )
-                for section, statistic_conf in STATISTIC_SECTIONS_CONF.items()
-                if (statistic_table := results_data.get(section)) is not None
-            ]
+            return statistic_dimensions
 
-        measurement_id = random_uuid_str()
         analytes = []
         calculated_data = []
-        for analyte in [
+        analyte_keys = [
             key for key in count_data.series.index if key not in metadata_keys
-        ]:
+        ]
+        for analyte in analyte_keys:
             analytes.append(
                 Analyte(
                     identifier=(analyte_identifier := random_uuid_str()),
@@ -217,14 +242,21 @@ def get_statistic_dimensions(analyte: str) -> list[StatisticDimension]:
                 )
             )
 
-            calculated_data.extend(
-                [
+            for section_name, unit in CALCULATED_DATA_SECTIONS.items():
+                calculated_data_section = results_data.get(section_name, pd.DataFrame())
+                # Some sections don't include the location as an index, in those cases we cannot associate
+                # the calculated data with the measurement.
+                if location not in calculated_data_section.index:
+                    continue
+                raw_value = calculated_data_section.at[location, analyte]
+                if error := get_statistics_error(raw_value, analyte, section_name):
+                    errors.append(error)
+                calculated_data_id = random_uuid_str()
+                calculated_data.append(
                     CalculatedDocument(
-                        uuid=random_uuid_str(),
+                        uuid=calculated_data_id,
                         name=section_name,
-                        value=try_float(
-                            calculated_data_section.at[location, analyte], analyte
-                        ),
+                        value=try_non_nan_float_or_negative_zero(raw_value),
                         unit=unit,
                         data_sources=[
                             DataSource(
@@ -233,15 +265,10 @@ def get_statistic_dimensions(analyte: str) -> list[StatisticDimension]:
                             )
                         ],
                     )
-                    for section_name, unit in CALCULATED_DATA_SECTIONS.items()
-                    if section_name in results_data
-                    and location
-                    in (calculated_data_section := results_data[section_name]).index
-                ]
-            )
+                )
 
         return Measurement(
-            identifier=measurement_id,
+            identifier=measurement_identifier,
             sample_identifier=count_data[str, "Sample"],
             location_identifier=location_id,
             dilution_factor_setting=dilution_factor_setting,

src/allotropy/parsers/utils/values.py

@@ -12,6 +12,7 @@
     TStatisticDatumRole,
 )
 from allotropy.exceptions import AllotropeConversionError, AllotropyParserError
+from allotropy.parsers.constants import NEGATIVE_ZERO
 from allotropy.parsers.utils.units import get_quantity_class
 
 PrimitiveValue = str | int | float
@@ -90,6 +91,11 @@ def try_non_nan_float(value: str) -> float:
     return float_value
 
 
+def try_non_nan_float_or_negative_zero(value: str) -> float:
+    float_value = try_non_nan_float_or_none(value)
+    return NEGATIVE_ZERO if float_value is None else float_value
+
+
 def try_int_or_nan(value: str | int | None) -> int | InvalidJsonFloat:
     if isinstance(value, int):
         return value