Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""
if autosomes_only:
mt = filter_to_autosomes(mt)
if bi_allelic_only:
mt = mt.filter_rows(bi_allelic_expr(mt))
sib_ht = relatedness_ht.filter(relatedness_ht[relationship_col] == SIBLINGS)
s_to_keep = sib_ht.aggregate(
hl.agg.explode(
lambda s: hl.agg.collect_as_set(s), [sib_ht[i_col].s, sib_ht[j_col].s]
),
_localize=False,
)
mt = mt.filter_cols(s_to_keep.contains(mt.s))
if "adj" not in mt.entry:
mt = annotate_adj(mt)
sib_stats_ht = mt.select_rows(
**generate_sib_stats_expr(
mt, sib_ht, i_col=i_col, j_col=j_col, strata={"raw": True, "adj": mt.adj},
)
).rows()
return sib_stats_ht
def filter_mt_to_trios(mt: hl.MatrixTable, fam_ht: hl.Table) -> hl.MatrixTable:
"""
Filters a MatrixTable to a set of trios in `fam_ht` and annotates with adj.
:param mt: A Matrix Table to filter to only trios
:param fam_ht: A Table of trios to filter to, loaded using `hl.import_fam`
:return: A MT filtered to trios and adj annotated
"""
# Filter MT to samples present in any of the trios
fam_ht = fam_ht.annotate(fam_members=[fam_ht.id, fam_ht.pat_id, fam_ht.mat_id])
fam_ht = fam_ht.explode("fam_members", name="s")
fam_ht = fam_ht.key_by("s").select().distinct()
mt = mt.filter_cols(hl.is_defined(fam_ht[mt.col_key]))
if "adj" not in mt.entry:
mt = annotate_adj(mt)
return mt
def filter_to_adj(mt: hl.MatrixTable) -> hl.MatrixTable:
"""
Filter genotypes to adj criteria
"""
if "adj" not in list(mt.entry):
mt = annotate_adj(mt)
mt = mt.filter_entries(mt.adj)
return mt.drop(mt.adj)