Skip to content
5 changes: 5 additions & 0 deletions src/allotropy/allotrope/converter.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from collections.abc import Callable, Mapping, Sequence
from dataclasses import asdict, field, fields, is_dataclass, make_dataclass, MISSING
from enum import Enum
import keyword
from types import GenericAlias, UnionType
from typing import (
Any,
Expand Down Expand Up @@ -164,6 +165,8 @@ def add_custom_information_document(

def _convert_model_key_to_dict_key(key: str) -> str:
key = SPECIAL_KEYS.get(key, key)
if key.startswith("_KW"):
key = key[3:]
if key.startswith("___") and key[3].isdigit():
key = key[3:]
for dict_val, model_val in DICT_KEY_TO_MODEL_KEY_REPLACEMENTS.items():
Expand All @@ -173,6 +176,8 @@ def _convert_model_key_to_dict_key(key: str) -> str:

def _convert_dict_to_model_key(key: str) -> str:
key = SPECIAL_KEYS_INVERSE.get(key, key)
if keyword.iskeyword(key):
key = f"_KW{key}"
if key[0].isdigit():
key = f"___{key}"
for dict_val, model_val in DICT_KEY_TO_MODEL_KEY_REPLACEMENTS.items():
Expand Down
42 changes: 37 additions & 5 deletions src/allotropy/parsers/ctl_immunospot/ctl_immunospot_reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -202,8 +202,6 @@ def _read_plate_data_7_0_38(self, reader: CsvReader) -> list[pd.DataFrame]:
return plates

def _read_header(self, reader: CsvReader) -> SeriesData:
lines = [line.strip() for line in reader.pop_until_empty(EMPTY_STR_OR_CSV_LINE)]

def fix_line(line: str) -> str:
# Add missing key for file path line.
if line.endswith(".txt") or line.endswith(".xls"):
Expand All @@ -213,13 +211,47 @@ def fix_line(line: str) -> str:
line = f"Review Date: {date_str}"
return line.strip()

lines = [fix_line(line) for raw_line in lines for line in raw_line.split(";")]
def split_multi_key_line(line: str) -> list[str]:
"""Split lines that contain multiple key-value pairs into separate lines."""
line = line.strip(' "') # Clean up quotes and whitespace

# Case 1: Parenthetical format: (Auto Areas: Estimated, Manual Areas: Normalized)
if line.startswith("(") and line.endswith(")") and "," in line:
return [pair.strip() for pair in line[1:-1].split(",")]

# Case 2: Tab-separated format: Assay: 1\t\t\t\t\t\tEdge Compensation Level: 1.0
if "\t" in line and line.count(":") > 1:
return [
part.strip()
for part in line.split("\t")
if part.strip() and ":" in part
]

return [line]

def process_lines(raw_lines: list[str]) -> list[str]:
"""Process a list of raw lines and return split/fixed lines."""
processed = []
for raw_line in raw_lines:
for line in raw_line.split(";"):
fixed_line = fix_line(line)
# Only process lines that have colons (key: value format)
if ":" in fixed_line:
processed.extend(split_multi_key_line(fixed_line))
return processed

# Process initial header lines
lines = [line.strip() for line in reader.pop_until_empty(EMPTY_STR_OR_CSV_LINE)]
all_lines = process_lines(lines)

# Process additional header lines if they exist
reader.drop_empty(EMPTY_STR_OR_CSV_LINE)
if ":" in (reader.get() or ""):
lines.extend(list(reader.pop_until_empty(EMPTY_STR_OR_CSV_LINE)))
additional_lines = list(reader.pop_until_empty(EMPTY_STR_OR_CSV_LINE))
all_lines.extend(process_lines(additional_lines))

df = read_csv(
StringIO("\n".join(lines)),
StringIO("\n".join(all_lines)),
sep=r"^([^:]+):\s+",
header=None,
engine="python",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,18 @@ def _create_measurement(
well_plate_identifier: str,
plate_data: dict[str, pd.DataFrame],
histograms: dict[str, tuple[list[float], list[float]]],
header_data: dict[str, float | str | None],
) -> Measurement:
location_identifier = f"{well_row}{well_col}"
data_processing_document = {
"Min. SpotSize": header_data.pop("Min. SpotSize", None),
"Max. SpotSize": header_data.pop("Max. SpotSize", None),
"Spot Separation": header_data.pop("Spot Separation", None),
}
has_data = any(
value is not None and str(value).strip() != ""
for value in data_processing_document.values()
)
return Measurement(
type_=MeasurementType.OPTICAL_IMAGING,
device_type=constants.DEVICE_TYPE,
Expand All @@ -45,6 +55,7 @@ def _create_measurement(
detection_type=constants.DETECTION_TYPE,
processed_data=ProcessedData(
identifier=random_uuid_str(),
data_processing_document=data_processing_document if has_data else None,
features=[
ImageFeature(
identifier=random_uuid_str(),
Expand Down Expand Up @@ -77,6 +88,7 @@ def _create_measurement(
]
if histograms and location_identifier in histograms
else None,
device_control_custom_info=header_data,
)


Expand All @@ -94,16 +106,28 @@ def create_measurement_groups(
round_to_nearest_well_count(first_plate.size),
f"Unable to determine valid plate count from dataframe of size: {first_plate.size}",
)
measurement_time = header[str, ("Counted", "Review Date")]
analyst = header[str, "Authenticated user"]
custom_info = {
"Assay": header.get(str, "Assay"),
}
header_data = header.get_unread()
return [
MeasurementGroup(
plate_well_count=plate_well_count,
measurement_time=header[str, ("Counted", "Review Date")],
analyst=header[str, "Authenticated user"],
measurement_time=measurement_time,
analyst=analyst,
measurements=[
_create_measurement(
row, col, well_plate_identifier, plate_data, histograms
row,
col,
well_plate_identifier,
plate_data,
histograms,
header_data.copy(),
)
],
custom_info=custom_info,
)
for row in first_plate.index
for col in first_plate.columns
Expand Down
1 change: 1 addition & 0 deletions src/allotropy/parsers/utils/pandas.py
Original file line number Diff line number Diff line change
Expand Up @@ -279,6 +279,7 @@ def _key_matches(self, match_key: str, key: str) -> bool:
# "++" can cause re.compile to fail. Since it is never a valid regex expression
# it is safe to escape it to prevent the error.
match_key = match_key.replace("++", r"\+\+")

return bool(re.fullmatch(match_key, key))

def _get_matching_keys(self, key_or_keys: str | set[str]) -> set[str]:
Expand Down
Loading
Loading