Skip to content

mecfs_bio.build_system.task.genome_reference_harmonization.resolve_chromosome

Resolve one chromosome of summary statistics against the genome reference.

Every input row is returned. Alleles are oriented so that NEA is the plus-strand reference allele, allele-dependent statistics are flipped to match, and DROP_REASON_COL is null for rows to keep. The function holds one chromosome in memory.

Classes:

Functions:

Attributes:

ACTION_COL module-attribute 

ACTION_COL = '_action'

DROP_REASON_COL module-attribute 

DROP_REASON_COL = '_drop_reason'

PALINDROME_DECISION_COL module-attribute 

PALINDROME_DECISION_COL = '_palindrome_decision'

ROW_INDEX_COL module-attribute 

ROW_INDEX_COL = '_row_index'

ChromosomeContext

Attributes:

chrom instance-attribute 

chrom: int

fasta instance-attribute 

fasta: IndexedFasta

options instance-attribute 

options: GenomeReferenceHarmonizationOptions

rules instance-attribute 

rules: Mapping[str, FlipRule]

trusted instance-attribute 

trusted: bool

PanelLoader

Bases: Protocol

resolve_chromosome 

resolve_chromosome(
    rows: DataFrame,
    context: ChromosomeContext,
    load_panel: PanelLoader,
) -> pl.DataFrame
Source code in mecfs_bio/build_system/task/genome_reference_harmonization/resolve_chromosome.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def resolve_chromosome(
    rows: pl.DataFrame, context: ChromosomeContext, load_panel: PanelLoader
) -> pl.DataFrame:
    output_columns = [*rows.columns, DROP_REASON_COL]
    frame = prepare_alleles(rows).with_row_index(ROW_INDEX_COL)
    invalid = frame.filter(~valid_alleles_expr()).with_columns(
        pl.lit(DROP_INVALID_ALLELE, dtype=pl.String).alias(DROP_REASON_COL)
    )
    valid = classify_alleles(
        frame.filter(valid_alleles_expr()),
        fasta=context.fasta,
        chrom=context.chrom,
        max_gather_bytes=context.options.max_gather_bytes,
    ).with_columns(_base_action_expr(), _base_drop_reason_expr(trusted=context.trusted))
    panel = None if context.trusted else load_panel(_panel_positions(valid))
    if panel is not None:
        valid = _apply_ambiguous_indel_rules(valid, panel, context)
    valid = _apply_allele_actions(valid, context.rules)
    if panel is not None:
        valid = _apply_palindrome_strand_rules(valid, panel, context)
    return (
        pl.concat(
            [
                valid.select(ROW_INDEX_COL, *output_columns),
                invalid.select(ROW_INDEX_COL, *output_columns),
            ]
        )
        .sort(ROW_INDEX_COL)
        .drop(ROW_INDEX_COL)
    )