feat: Agilent Gen5 - Add Conc/Dil to Sample document custom fields

src/allotropy/parsers/agilent_gen5/agilent_gen5_parser.py

@@ -13,6 +13,7 @@
     create_metadata,
     create_results,
     create_spectrum_results,
+    get_concentrations,
     get_identifiers,
     get_kinetic_measurements,
     get_results_section,
@@ -49,6 +50,7 @@ def create_data(self, named_file_contents: NamedFileContents) -> Data:
         kinetic_data = KineticData.create(reader.sections["Procedure Details"])
 
         sample_identifiers = get_identifiers(reader.sections.get("Layout"))
+        concentration_values = get_concentrations(reader.sections.get("Layout"))
         actual_temperature = get_temperature(reader.sections.get("Actual Temperature"))
         kinetic_result = get_kinetic_measurements(reader.sections.get("Time"))
         kinetic_measurements, kinetic_elapsed_time, kinetic_errors = kinetic_result or (
@@ -73,6 +75,7 @@ def create_data(self, named_file_contents: NamedFileContents) -> Data:
                 sample_identifiers=sample_identifiers,
                 actual_temperature=actual_temperature,
                 results_section=results_section,
+                concentration_values=concentration_values,
             )
 
             if not wavelength_measurements:
@@ -95,6 +98,7 @@ def create_data(self, named_file_contents: NamedFileContents) -> Data:
                 kinetic_measurements,
                 kinetic_elapsed_time,
                 kinetic_errors,
+                concentration_values,
             )
         else:
             measurement_groups, calculated_data = create_results(
@@ -103,6 +107,7 @@ def create_data(self, named_file_contents: NamedFileContents) -> Data:
                 read_data,
                 sample_identifiers,
                 actual_temperature,
+                concentration_values,
             )
 
         if not measurement_groups:

src/allotropy/parsers/agilent_gen5/agilent_gen5_structure.py

@@ -650,6 +650,34 @@ def is_results(section: list[str]) -> bool:
     return None
 
 
+def get_concentrations(layout_lines: list[str] | None) -> dict[str, float | None]:
+    """Extract concentration/dilution values from the Layout section.
+
+    Returns a dictionary mapping well positions to their concentration/dilution values.
+    """
+    if not layout_lines:
+        return {}
+
+    # Create dataframe from tabular data and forward fill empty values in index
+    data = read_csv(StringIO("\n".join(layout_lines[1:])), sep="\t")
+    data = data.set_index(data.index.to_series().ffill(axis="index").values)
+
+    concentrations = {}
+    for row_name, row in data.iterrows():
+        label = row.iloc[-1]
+        if label == "Conc/Dil":
+            for col_index, col in enumerate(row.iloc[:-1]):
+                well_pos = f"{row_name}{col_index + 1}"
+                # Convert to float if possible
+                if not pd.isna(col):
+                    concentration_value = try_float_or_none(col)
+                    concentrations[well_pos] = concentration_value
+                else:
+                    concentrations[well_pos] = None
+
+    return concentrations
+
+
 def get_identifiers(layout_lines: list[str] | None) -> dict[str, str]:
     if not layout_lines:
         return {}
@@ -759,6 +787,7 @@ def create_results(
     read_data: list[ReadData],
     sample_identifiers: dict[str, str],
     actual_temperature: float | None,
+    concentration_values: dict[str, float | None] | None = None,
 ) -> tuple[list[MeasurementGroup], list[CalculatedDocument]]:
     if result_lines[0].strip() != "Results":
         msg = f"Expected the first line of the results section '{result_lines[0]}' to be 'Results'."
@@ -827,6 +856,9 @@ def create_results(
                     sample_identifiers.get(well_position),
                     actual_temperature,
                     error_documents=error_documents_per_well.get(well_position),
+                    concentration_value=concentration_values.get(well_position)
+                    if concentration_values
+                    else None,
                 )
                 for measurement in measurements
             ],
@@ -868,6 +900,7 @@ def create_kinetic_results(
     kinetic_measurements: dict[str, list[float | None]],
     kinetic_elapsed_time: list[float],
     kinetic_errors: dict[str, list[ErrorDocument]] | None = None,
+    concentration_values: dict[str, float | None] | None = None,
 ) -> tuple[list[MeasurementGroup], list[CalculatedDocument]]:
     if result_lines[0].strip() != "Results":
         msg = f"Expected the first line of the results section '{result_lines[0]}' to be 'Results'."
@@ -919,6 +952,9 @@ def create_kinetic_results(
                     kinetic_measurements[well_position],
                     kinetic_elapsed_time,
                     error_documents_per_well.get(well_position, []),
+                    concentration_value=concentration_values.get(well_position)
+                    if concentration_values
+                    else None,
                 )
             ],
         )
@@ -1080,6 +1116,7 @@ def create_spectrum_results(
     sample_identifiers: dict[str, str],
     actual_temperature: float | None,
     results_section: list[str] | None = None,
+    concentration_values: dict[str, float | None] | None = None,
 ) -> tuple[list[MeasurementGroup], list[CalculatedDocument]]:
     if not wavelengths_sections:
         return [], []
@@ -1240,6 +1277,11 @@ def create_spectrum_results(
             else None,
             detector_gain_setting=filter_set.gain if filter_set else None,
             error_document=error_documents_by_well.get(well_position),
+            sample_custom_info={
+                "Conc/Dil": concentration_values.get(well_position)
+                if concentration_values
+                else None,
+            },
             analytical_method_identifier=header_data.protocol_file_path
             if header_data.protocol_file_path
             else None,
@@ -1298,6 +1340,7 @@ def _create_measurement(
     kinetic_measurements: list[float | None] | None = None,
     kinetic_elapsed_time: list[float] | None = None,
     error_documents: list[ErrorDocument] | None = None,
+    concentration_value: float | None = None,
 ) -> Measurement:
     if read_data.read_mode == ReadMode.ABSORBANCE and not kinetic_data:
         measurement_type = MeasurementType.ULTRAVIOLET_ABSORBANCE
@@ -1408,7 +1451,8 @@ def _create_measurement(
             ),
         },
         sample_custom_info={
-            "Plate Number": header_data.additional_data.pop("Plate Number", None)
+            "Plate Number": header_data.additional_data.pop("Plate Number", None),
+            "Conc/Dil": concentration_value,
         },
         measurement_custom_info=header_data.additional_data,
         analytical_method_identifier=header_data.protocol_file_path