Skip to content

Commit 938a1a2

Browse files
fix: Support new PkN column format in Stunner parser (#1205)
## Summary - Stunner parser now recognizes both old (`Peak N ...`) and new (`PkN ...`) column naming conventions for peak data - PkOI (Peak of Interest) columns are consumed but not emitted as a separate peak entry, eliminating the duplicate column issue customers reported - Peak data (mean diameter, intensity area, mass area, est. MW, rayleigh ratio) is now properly structured under `peak_list` in the output instead of leaking as raw columns in `custom_information_document` ## Context Customer reported duplicate columns in connector output — e.g. two "Pk1 Intensity Area (%)" columns where one was empty. Root cause: newer Stunner exports use `PkN` prefix instead of `Peak N`, and `PkOI` (Peak of Interest) duplicates the primary peak's data. ## Test plan - [x] All 32 existing Stunner/Lunatic tests pass - [x] All 10 Lunatic discovery tests pass - [x] Verified customer file: 60/85 measurements correctly produce structured peak data - [x] Verified no pk*/pkoi columns leak into measurement_custom_info - [x] Lint clean Generated with Claude Code --------- Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
1 parent 75870b0 commit 938a1a2

3 files changed

Lines changed: 1718 additions & 29 deletions

File tree

src/allotropy/parsers/unchained_labs_lunatic_stunner/unchained_labs_lunatic_stunner_structure.py

Lines changed: 54 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -56,46 +56,71 @@
5656
)
5757

5858

59+
def _extract_peak_info(
60+
well_plate_data: SeriesData, prefix: str, index: str
61+
) -> dict[str, Any] | None:
62+
"""Extract a single peak's data using the given column prefix (e.g. 'peak 1' or 'pk1')."""
63+
peak_info: dict[str, Any] = {}
64+
65+
# Old format: "peak N mean dia (nm)", "peak N intensity (%)", etc.
66+
# New format: "pkN intensity mean dia. (nm)", "pkN intensity area (%)", etc.
67+
field_mappings: list[tuple[str, list[str]]] = [
68+
(
69+
"peak mean diameter",
70+
[f"{prefix} mean dia (nm)", f"{prefix} intensity mean dia. (nm)"],
71+
),
72+
("peak mode diameter", [f"{prefix} mode dia (nm)"]),
73+
("peak est. MW", [f"{prefix} est. mw (kda)"]),
74+
("peak intensity", [f"{prefix} intensity (%)", f"{prefix} intensity area (%)"]),
75+
("peak mass", [f"{prefix} mass (%)", f"{prefix} mass area (%)"]),
76+
("peak mass mean diameter", [f"{prefix} mass mean dia. (nm)"]),
77+
("peak rayleigh ratio", [f"{prefix} rayleigh ratio r (1/km)"]),
78+
("peak diffusion coefficient", [f"{prefix} diffusion coefficient (um^2/s)"]),
79+
]
80+
81+
for output_key, candidate_columns in field_mappings:
82+
for col in candidate_columns:
83+
val = well_plate_data.get(float, col)
84+
if val is not None:
85+
peak_info[output_key] = val
86+
break
87+
88+
peak_info["peak index"] = index
89+
90+
if len(peak_info) > 1:
91+
return peak_info
92+
return None
93+
94+
5995
def _extract_peak_data(well_plate_data: SeriesData) -> list[dict[str, Any]]:
6096
"""Extract peak data from well plate data and return as list of peak info dictionaries."""
6197
peak_data = []
6298

63-
peak_pattern = re.compile(r"peak (\d+) ")
64-
peak_numbers = set()
99+
# Detect which format is used by scanning column names
100+
old_pattern = re.compile(r"peak (\d+) ")
101+
new_pattern = re.compile(r"pk(\d+) ")
102+
peak_numbers: set[int] = set()
103+
uses_new_format = False
65104

66105
for column in well_plate_data.series.index:
67-
match = peak_pattern.match(str(column).lower())
68-
if match:
106+
col_lower = str(column).lower()
107+
if match := old_pattern.match(col_lower):
69108
peak_numbers.add(int(match.group(1)))
109+
elif match := new_pattern.match(col_lower):
110+
peak_numbers.add(int(match.group(1)))
111+
uses_new_format = True
70112

71113
for peak_num in sorted(peak_numbers):
72-
peak_info: dict[str, Any] = {}
73-
74-
mean_dia = well_plate_data.get(float, f"peak {peak_num} mean dia (nm)")
75-
if mean_dia is not None:
76-
peak_info["peak mean diameter"] = mean_dia
77-
78-
mode_dia = well_plate_data.get(float, f"peak {peak_num} mode dia (nm)")
79-
if mode_dia is not None:
80-
peak_info["peak mode diameter"] = mode_dia
81-
82-
est_mw = well_plate_data.get(float, f"peak {peak_num} est. mw (kda)")
83-
if est_mw is not None:
84-
peak_info["peak est. MW"] = est_mw
85-
86-
intensity = well_plate_data.get(float, f"peak {peak_num} intensity (%)")
87-
if intensity is not None:
88-
peak_info["peak intensity"] = intensity
89-
90-
mass = well_plate_data.get(float, f"peak {peak_num} mass (%)")
91-
if mass is not None:
92-
peak_info["peak mass"] = mass
93-
94-
peak_info["peak index"] = str(peak_num)
95-
96-
if len(peak_info) > 1:
114+
prefix = f"pk{peak_num}" if uses_new_format else f"peak {peak_num}"
115+
if peak_info := _extract_peak_info(well_plate_data, prefix, str(peak_num)):
97116
peak_data.append(peak_info)
98117

118+
# Read "peak of interest" / "pkoi" columns to prevent them from leaking
119+
# into custom_info, but don't add as a separate peak entry since it
120+
# duplicates one of the numbered peaks.
121+
poi_prefix = "pkoi" if uses_new_format else "peak of interest"
122+
_extract_peak_info(well_plate_data, poi_prefix, "OI")
123+
99124
return peak_data
100125

101126

0 commit comments

Comments
 (0)