Skip to content

Commit

Permalink
chore: Refactor Appbio Quantstuido calculated documents yield logic (#…
Browse files Browse the repository at this point in the history 
…704)

This does not change any functionality, just makes this code a bit
easier to update by moving the yielding logic to a helper.
  • Loading branch information
@nathan-stender
nathan-stender authored Oct 8, 2024
1 parent 43a0905 commit 3c47027
Show file tree
Hide file tree
Showing 2 changed files with 115 additions and 186 deletions.
175 changes: 68 additions & 107 deletions src/allotropy/parsers/appbio_quantstudio/appbio_quantstudio_calculated_documents.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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],
Expand All @@ -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(
Expand All @@ -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(
Expand All @@ -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(
Expand Down
126 changes: 47 additions & 79 deletions ...uantstudio_designandanalysis/appbio_quantstudio_designandanalysis_calculated_documents.py
View file
Original file line number Diff line number Diff line change
@@ -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,
Expand Down Expand Up @@ -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(
Expand All @@ -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)

0 comments on commit 3c47027

Please sign in to comment.