Skip to content
Open
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
10 changes: 8 additions & 2 deletions src/pmotools/pmo_engine/pmo_reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -394,10 +394,11 @@ def combine_multiple_pmos(pmos: list[dict]):
pmo["bioinformatics_run_info"]
):
bioinformatics_run_info_copy = copy.deepcopy(bioinformatics_run_info)
# remap using the run's own methods id, not its position index
bioinformatics_run_info_copy[
"bioinformatics_methods_id"
] = bioinformatics_methods_info_old_index_key[pmo_index][
bioinformatics_run_info_index
bioinformatics_run_info_copy["bioinformatics_methods_id"]
]
if "bioinformatics_run_info" not in pmo_out:
pmo_out["bioinformatics_run_info"] = []
Expand Down Expand Up @@ -501,8 +502,13 @@ def combine_multiple_pmos(pmos: list[dict]):
new_mhaps_target_index = len(
pmo_out["representative_microhaplotypes"]["targets"]
)
new_rep_target = copy.deepcopy(representative_microhaplotypes)
# remap the new target's target_id to the combined target_info
new_rep_target["target_id"] = target_info_old_index_key[pmo_index][
new_rep_target["target_id"]
]
pmo_out["representative_microhaplotypes"]["targets"].append(
copy.deepcopy(representative_microhaplotypes)
new_rep_target
)
representative_microhaplotypes_old_index_key[pmo_index][
representative_microhaplotypes_index
Expand Down
67 changes: 67 additions & 0 deletions tests/test_pmo_engine/test_pmo_reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,73 @@ def test_combine_multiple_pmos_fail_dup_library_sample_names(self):
)
self.assertRaises(Exception, PMOReader.combine_multiple_pmos, pmo_data_list_2)

def test_combine_multiple_pmos_remaps_new_rep_target_id(self):
# a representative-microhaplotype target newly added from a
# non-first PMO must have its target_id remapped to the combined
# target_info (previously it kept its local target_id), this test
# ensures this is tested as it was previously untested (2026-06-29)
import pandas as pd
from pmotools.pmo_builder.mhap_table_to_pmo import mhap_table_to_pmo
from pmotools.pmo_builder.panel_information_to_pmo import (
panel_info_table_to_pmo,
)
from pmotools.pmo_builder.merge_to_pmo import merge_to_pmo

def build(libs, targets, seqs, panel_name, panel_targets):
mhap_info = mhap_table_to_pmo(
pd.DataFrame(
{
"library_sample_name": libs,
"target_name": targets,
"seq": seqs,
"reads": [10] * len(libs),
}
)
)
panel_info = panel_info_table_to_pmo(
pd.DataFrame(
{
"target_name": panel_targets,
"fwd_primer": [
"A" * (i + 4) for i in range(len(panel_targets))
],
"rev_primer": [
"C" * (i + 4) for i in range(len(panel_targets))
],
}
),
panel_name,
)
return merge_to_pmo(mhap_info=mhap_info, panel_target_info=panel_info)

pmo_a = build(
["S1", "S2"], ["t1", "t2"], ["AAA", "GGG"], "panelA", ["t1", "t2"]
)
# pmo_b introduces a brand-new target t3
pmo_b = build(
["S3", "S4"], ["t1", "t3"], ["AAA", "CCC"], "panelB", ["t1", "t3"]
)

combined = PMOReader.combine_multiple_pmos([pmo_a, pmo_b])
self.assertEqual(
[t["target_name"] for t in combined["target_info"]],
["t1", "t2", "t3"],
)
# every rep target's target_id resolves to the right combined target
for rep in combined["representative_microhaplotypes"]["targets"]:
self.assertLess(rep["target_id"], len(combined["target_info"]))
t3_reps = [
rep
for rep in combined["representative_microhaplotypes"]["targets"]
if combined["target_info"][rep["target_id"]]["target_name"] == "t3"
]
self.assertEqual(len(t3_reps), 1)
# validate against schema (minimal builder PMO targets v1.1.0)
checker = PMOChecker(
load_schema("portable_microhaplotype_object_v1.1.0.schema.json")
)
checker.validate_pmo_json(combined)

def test_combine_multiple_pmos_fail_for_combine_only_one_file(self):
# will fail for only having 1 PMO
pmo_data_list_2 = PMOReader.read_in_pmos(
Expand Down
Loading