From 3c47027d203bc4cb781301b8ad284bccb5086182 Mon Sep 17 00:00:00 2001 From: Nathan Stender Date: Tue, 8 Oct 2024 08:48:35 -0400 Subject: [PATCH] chore: Refactor Appbio Quantstuido calculated documents yield logic (#704) This does not change any functionality, just makes this code a bit easier to update by moving the yielding logic to a helper. --- ...appbio_quantstudio_calculated_documents.py | 175 +++++++----------- ..._designandanalysis_calculated_documents.py | 126 +++++-------- 2 files changed, 115 insertions(+), 186 deletions(-) diff --git a/src/allotropy/parsers/appbio_quantstudio/appbio_quantstudio_calculated_documents.py b/src/allotropy/parsers/appbio_quantstudio/appbio_quantstudio_calculated_documents.py index 1b2e5f45c..c1a0058aa 100644 --- a/src/allotropy/parsers/appbio_quantstudio/appbio_quantstudio_calculated_documents.py +++ b/src/allotropy/parsers/appbio_quantstudio/appbio_quantstudio_calculated_documents.py @@ -586,6 +586,14 @@ def build_efficiency( ) +def yield_documents( + calc_docs: list[CalculatedDocument | None], +) -> Iterator[CalculatedDocument]: + for calc_doc in calc_docs: + if calc_doc: + yield from calc_doc.iter_struct() + + def iter_comparative_ct_calc_docs( view_st_data: ViewData[WellItem], view_tr_data: ViewData[WellItem], @@ -594,39 +602,29 @@ def iter_comparative_ct_calc_docs( ) -> Iterator[CalculatedDocument]: # Quantity, Quantity Mean, Quantity SD, Ct Mean, Ct SD, Delta Ct Mean, # Delta Ct SE, Delta Delta Ct, RQ, RQ min, RQ max, Amplification score, Cq confidence + calc_docs: list[CalculatedDocument | None] = [] for sample, target in view_st_data.iter_keys(): for well_item in view_st_data.get_leaf_item(sample, target): - if calc_doc := build_quantity(view_tr_data, target, well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_amp_score(well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_cq_conf(well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_quantity_mean(view_st_data, view_tr_data, sample, target): - yield from calc_doc.iter_struct() + calc_docs.append(build_quantity(view_tr_data, target, well_item)) + calc_docs.append(build_amp_score(well_item)) + calc_docs.append(build_cq_conf(well_item)) - if calc_doc := build_quantity_sd(view_st_data, view_tr_data, sample, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_ct_sd(view_st_data, sample, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_delta_ct_se(view_st_data, sample, target, r_target): - yield from calc_doc.iter_struct() + calc_docs.append( + build_quantity_mean(view_st_data, view_tr_data, sample, target) + ) + calc_docs.append(build_quantity_sd(view_st_data, view_tr_data, sample, target)) + calc_docs.append(build_ct_sd(view_st_data, sample, target)) + calc_docs.append(build_delta_ct_se(view_st_data, sample, target, r_target)) if target != r_target: - if calc_doc := build_rq_min( - view_st_data, sample, target, r_sample, r_target - ): - yield from calc_doc.iter_struct() + calc_docs.append( + build_rq_min(view_st_data, sample, target, r_sample, r_target) + ) + calc_docs.append( + build_rq_max(view_st_data, sample, target, r_sample, r_target) + ) - if calc_doc := build_rq_max( - view_st_data, sample, target, r_sample, r_target - ): - yield from calc_doc.iter_struct() + yield from yield_documents(calc_docs) def iter_standard_curve_calc_docs( @@ -635,41 +633,27 @@ def iter_standard_curve_calc_docs( ) -> Iterator[CalculatedDocument]: # Quantity, Quantity Mean, Quantity SD, Ct Mean, Ct SD, Y-Intercept, # R(superscript 2), Slope, Efficiency, Amplification score, Cq confidence + calc_docs: list[CalculatedDocument | None] = [] for sample, target in view_st_data.iter_keys(): for well_item in view_st_data.get_leaf_item(sample, target): - if calc_doc := build_quantity(view_tr_data, target, well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_amp_score(well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_cq_conf(well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_quantity_mean(view_st_data, view_tr_data, sample, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_quantity_sd(view_st_data, view_tr_data, sample, target): - yield from calc_doc.iter_struct() + calc_docs.append(build_quantity(view_tr_data, target, well_item)) + calc_docs.append(build_amp_score(well_item)) + calc_docs.append(build_cq_conf(well_item)) - if calc_doc := build_ct_mean(view_st_data, sample, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_ct_sd(view_st_data, sample, target): - yield from calc_doc.iter_struct() + calc_docs.append( + build_quantity_mean(view_st_data, view_tr_data, sample, target) + ) + calc_docs.append(build_quantity_sd(view_st_data, view_tr_data, sample, target)) + calc_docs.append(build_ct_mean(view_st_data, sample, target)) + calc_docs.append(build_ct_sd(view_st_data, sample, target)) for target in view_tr_data.data: - if calc_doc := build_y_intercept(view_tr_data, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_r_squared(view_tr_data, target): - yield from calc_doc.iter_struct() + calc_docs.append(build_y_intercept(view_tr_data, target)) + calc_docs.append(build_r_squared(view_tr_data, target)) + calc_docs.append(build_slope(view_tr_data, target)) + calc_docs.append(build_efficiency(view_tr_data, target)) - if calc_doc := build_slope(view_tr_data, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_efficiency(view_tr_data, target): - yield from calc_doc.iter_struct() + yield from yield_documents(calc_docs) def iter_relative_standard_curve_calc_docs( @@ -679,73 +663,50 @@ def iter_relative_standard_curve_calc_docs( # Quantity, Quantity Mean, Quantity SD, Ct Mean, Ct SD, RQ, RQ min, # RQ max, Y-Intercept, R(superscript 2), Slope, Efficiency, # Amplification score, Cq confidence + calc_docs: list[CalculatedDocument | None] = [] for sample, target in view_st_data.iter_keys(): for well_item in view_st_data.get_leaf_item(sample, target): - if calc_doc := build_quantity(view_tr_data, target, well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_amp_score(well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_cq_conf(well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_quantity_mean(view_st_data, view_tr_data, sample, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_quantity_sd(view_st_data, view_tr_data, sample, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_ct_mean(view_st_data, sample, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_ct_sd(view_st_data, sample, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_relative_rq_min( - view_st_data, view_tr_data, sample, target - ): - yield from calc_doc.iter_struct() + calc_docs.append(build_quantity(view_tr_data, target, well_item)) + calc_docs.append(build_amp_score(well_item)) + calc_docs.append(build_cq_conf(well_item)) - if calc_doc := build_relative_rq_max( - view_st_data, view_tr_data, sample, target - ): - yield from calc_doc.iter_struct() + calc_docs.append( + build_quantity_mean(view_st_data, view_tr_data, sample, target) + ) + calc_docs.append(build_quantity_sd(view_st_data, view_tr_data, sample, target)) + calc_docs.append(build_ct_mean(view_st_data, sample, target)) + calc_docs.append(build_ct_sd(view_st_data, sample, target)) + calc_docs.append( + build_relative_rq_min(view_st_data, view_tr_data, sample, target) + ) + calc_docs.append( + build_relative_rq_max(view_st_data, view_tr_data, sample, target) + ) for target in view_tr_data.data: - if calc_doc := build_y_intercept(view_tr_data, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_r_squared(view_tr_data, target): - yield from calc_doc.iter_struct() + calc_docs.append(build_y_intercept(view_tr_data, target)) + calc_docs.append(build_r_squared(view_tr_data, target)) + calc_docs.append(build_slope(view_tr_data, target)) + calc_docs.append(build_efficiency(view_tr_data, target)) - if calc_doc := build_slope(view_tr_data, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_efficiency(view_tr_data, target): - yield from calc_doc.iter_struct() + yield from yield_documents(calc_docs) def iter_presence_absence_calc_docs( view_data: ViewData[WellItem], ) -> Iterator[CalculatedDocument]: # Rn Mean, Rn SD, Amplification score, Cq confidence + calc_docs: list[CalculatedDocument | None] = [] for sample, target in view_data.iter_keys(): for well_item in view_data.get_leaf_item(sample, target): - if calc_doc := build_quantity(None, target, well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_amp_score(well_item): - yield from calc_doc.iter_struct() + calc_docs.append(build_quantity(None, target, well_item)) + calc_docs.append(build_amp_score(well_item)) + calc_docs.append(build_cq_conf(well_item)) - if calc_doc := build_cq_conf(well_item): - yield from calc_doc.iter_struct() + calc_docs.append(build_rn_mean(view_data, sample, target)) + calc_docs.append(build_rn_sd(view_data, sample, target)) - if calc_doc := build_rn_mean(view_data, sample, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_rn_sd(view_data, sample, target): - yield from calc_doc.iter_struct() + yield from yield_documents(calc_docs) def iter_calculated_data_documents( diff --git a/src/allotropy/parsers/appbio_quantstudio_designandanalysis/appbio_quantstudio_designandanalysis_calculated_documents.py b/src/allotropy/parsers/appbio_quantstudio_designandanalysis/appbio_quantstudio_designandanalysis_calculated_documents.py index 1a6aa3b9f..d53c02f64 100644 --- a/src/allotropy/parsers/appbio_quantstudio_designandanalysis/appbio_quantstudio_designandanalysis_calculated_documents.py +++ b/src/allotropy/parsers/appbio_quantstudio_designandanalysis/appbio_quantstudio_designandanalysis_calculated_documents.py @@ -1,6 +1,9 @@ from collections.abc import Iterator from functools import cache +from allotropy.parsers.appbio_quantstudio.appbio_quantstudio_calculated_documents import ( + yield_documents, +) from allotropy.parsers.appbio_quantstudio.views import ViewData from allotropy.parsers.appbio_quantstudio_designandanalysis.structure.generic.structure import ( WellItem, @@ -688,41 +691,27 @@ def iter_standard_curve_calc_docs( ) -> Iterator[CalculatedDocument]: # Quantity, Quantity Mean, Quantity SD, Ct Mean, Ct SD, Y-Intercept, # R(superscript 2), Slope, Efficiency, Amp score, Cq confidence + calc_docs: list[CalculatedDocument | None] = [] for sample, target in view_st_data.iter_keys(): for well_item in view_st_data.get_leaf_item(sample, target): - if calc_doc := build_quantity(view_tr_data, target, well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_amp_score(well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_cq_conf(well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_quantity_mean(view_st_data, view_tr_data, sample, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_quantity_sd(view_st_data, view_tr_data, sample, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_ct_mean(view_st_data, sample, target): - yield from calc_doc.iter_struct() + calc_docs.append(build_quantity(view_tr_data, target, well_item)) + calc_docs.append(build_amp_score(well_item)) + calc_docs.append(build_cq_conf(well_item)) - if calc_doc := build_ct_sd(view_st_data, sample, target): - yield from calc_doc.iter_struct() + calc_docs.append( + build_quantity_mean(view_st_data, view_tr_data, sample, target) + ) + calc_docs.append(build_quantity_sd(view_st_data, view_tr_data, sample, target)) + calc_docs.append(build_ct_mean(view_st_data, sample, target)) + calc_docs.append(build_ct_sd(view_st_data, sample, target)) for target in view_tr_data.data: - if calc_doc := build_y_intercept(view_tr_data, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_r_squared(view_tr_data, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_slope(view_tr_data, target): - yield from calc_doc.iter_struct() + calc_docs.append(build_y_intercept(view_tr_data, target)) + calc_docs.append(build_r_squared(view_tr_data, target)) + calc_docs.append(build_slope(view_tr_data, target)) + calc_docs.append(build_efficiency(view_tr_data, target)) - if calc_doc := build_efficiency(view_tr_data, target): - yield from calc_doc.iter_struct() + yield from yield_documents(calc_docs) def iter_relative_standard_curve_calc_docs( @@ -734,68 +723,47 @@ def iter_relative_standard_curve_calc_docs( # Quantity, Quantity Mean, Quantity SD, Ct Mean, Ct SD, RQ, RQ min, # RQ max, Y-Intercept, R(superscript 2), Slope, Efficiency, # Amp score, Cq confidence + calc_docs: list[CalculatedDocument | None] = [] for sample, target in view_st_data.iter_keys(): for well_item in view_st_data.get_leaf_item(sample, target): - if calc_doc := build_quantity(view_tr_data, target, well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_amp_score(well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_cq_conf(well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_ct_mean(view_st_data, sample, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_ct_sd(view_st_data, sample, target): - yield from calc_doc.iter_struct() - - if calc_doc := build_delta_ct_sd(view_st_data, sample, target, r_target): - yield from calc_doc.iter_struct() - - if calc_doc := build_delta_ct_se(view_st_data, sample, target, r_target): - yield from calc_doc.iter_struct() - - if calc_doc := build_relative_rq_min( - view_st_data, view_tr_data, sample, target - ): - yield from calc_doc.iter_struct() - - if calc_doc := build_relative_rq_max( - view_st_data, view_tr_data, sample, target - ): - yield from calc_doc.iter_struct() + calc_docs.append(build_quantity(view_tr_data, target, well_item)) + calc_docs.append(build_amp_score(well_item)) + calc_docs.append(build_cq_conf(well_item)) + + calc_docs.append(build_ct_mean(view_st_data, sample, target)) + calc_docs.append(build_ct_sd(view_st_data, sample, target)) + calc_docs.append(build_delta_ct_sd(view_st_data, sample, target, r_target)) + calc_docs.append(build_delta_ct_se(view_st_data, sample, target, r_target)) + calc_docs.append( + build_relative_rq_min(view_st_data, view_tr_data, sample, target) + ) + calc_docs.append( + build_relative_rq_min(view_st_data, view_tr_data, sample, target) + ) if target != r_target: - if calc_doc := build_rq_min( - view_st_data, sample, target, r_sample, r_target - ): - yield from calc_doc.iter_struct() + calc_docs.append( + build_rq_min(view_st_data, sample, target, r_sample, r_target) + ) + calc_docs.append( + build_rq_max(view_st_data, sample, target, r_sample, r_target) + ) - if calc_doc := build_rq_max( - view_st_data, sample, target, r_sample, r_target - ): - yield from calc_doc.iter_struct() + yield from yield_documents(calc_docs) def iter_presence_absence_calc_docs( view_data: ViewData[WellItem], ) -> Iterator[CalculatedDocument]: # Rn Mean, Rn SD, Amp score, Cq confidence + calc_docs: list[CalculatedDocument | None] = [] for sample, target in view_data.iter_keys(): for well_item in view_data.get_leaf_item(sample, target): - if calc_doc := build_quantity(None, target, well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_amp_score(well_item): - yield from calc_doc.iter_struct() - - if calc_doc := build_cq_conf(well_item): - yield from calc_doc.iter_struct() + calc_docs.append(build_quantity(None, target, well_item)) + calc_docs.append(build_amp_score(well_item)) + calc_docs.append(build_cq_conf(well_item)) - if calc_doc := build_rn_mean(view_data, sample, target): - yield from calc_doc.iter_struct() + calc_docs.append(build_rn_mean(view_data, sample, target)) + calc_docs.append(build_rn_sd(view_data, sample, target)) - if calc_doc := build_rn_sd(view_data, sample, target): - yield from calc_doc.iter_struct() + yield from yield_documents(calc_docs)