Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
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