h+ $dZddlZddlmZmZddlZddlmZddlZ ddl m Z ddl m Z mZmZmZddlmZmZe edde edd d e edd d e edd d e edd d d d eedfdeedfdeedfdeedfdeedff dZdee je jfdede jfdZdeee je jfdede jfdZdS)aPPost-processing utilities for Fi-NeMo hit calling results. This module provides functions for: - Collapsing overlapping hits based on similarity scores - Intersecting hit sets across multiple runs - Quality control and filtering operations The main operations are optimized using Numba for efficient processing of large hit datasets. N)ListUnion)ndarray)njit)Arrayuint32int32float32)FloatIntCT)readonly)cache chrom_idsz Nstartsends similaritiesreturnc|jd}tj|tj}dt dD}t |D]}||}||} ||} ||} |r4|d|| dfkr%t j||r|d|| dfk%|D]3\} } } | || k}|| xx|zcc<||xx| zcc<4t j||| |f|S)aIdentify primary hits among overlapping hits using a sweep line algorithm. This function uses a heap-based sweep line algorithm to efficiently identify the best hit (highest similarity) among sets of overlapping hits within each chromosome. Only one hit per overlapping group is marked as primary. Parameters ---------- chrom_ids : Int[ndarray, "N"] Chromosome identifiers for each hit, where N is the number of hits. Dtype should be uint32 for Numba compatibility. starts : Int[ndarray, "N"] Start positions of hits (adjusted for overlap computation). Dtype should be int32 for Numba compatibility. ends : Int[ndarray, "N"] End positions of hits (adjusted for overlap computation). Dtype should be int32 for Numba compatibility. similarities : Float[ndarray, "N"] Similarity scores used for selecting the best hit. Dtype should be float32 for Numba compatibility. Returns ------- Int[ndarray, "N"] Binary array where 1 indicates the hit is primary, 0 otherwise. Returns uint32 array for consistency with input types. Notes ----- This function is JIT-compiled with Numba for performance on large datasets. The algorithm maintains active intervals in a heap and resolves overlaps by keeping only the hit with the highest similarity score. The sweep line algorithm processes hits in order and maintains a heap of currently active intervals. When a new interval is encountered, it is compared against all overlapping intervals in the heap, and only the interval with the highest similarity score remains marked as primary. rdtypec`g|]+}tjdtjddf,S)r)nprr ).0_s z"_collapse_hits..Ns/ > > >RYq\\28A;; + > > >r)shaperonesrrangeheapqheappopheappush)rrrrnoutheapi chrom_new start_newend_newsim_newridxcmps r_collapse_hitsr1s?j A '!29 % % %C > >U1XX > > >D 1XX66aL 1I q'q/ tAw)Y!;;; M$    tAw)Y!;;;  IAq#L--C HHHOHHH FFF#g FFFF ti!45555 Jr hits_df overlap_fracc t|tjr|}|dd}|s4dt |D}|tjd |tj tjddztjdtjdz |z tj ztjddztjdtjdz |z tj z tjd  }n|tjd tjddztjdtjdz |z tj ztjddztjdtjdz |z tj z tjd  }| }|d d }|dd }|dd }|dd }t||||} |tj| tj } | S)aCollapse overlapping hits by selecting the best hit per overlapping group. This function identifies overlapping hits and marks only the highest-similarity hit as primary in each overlapping group. Overlap is determined by a fractional threshold based on the average length of the two hits being compared. Parameters ---------- hits_df : Union[pl.DataFrame, pl.LazyFrame] Hit data containing required columns: chr (or peak_id if no chr), start, end, hit_similarity. Will be collected to DataFrame if passed as LazyFrame. overlap_frac : float Overlap fraction threshold for considering hits as overlapping. For two hits with lengths x and y, minimum overlap = overlap_frac * (x + y) / 2. Must be between 0 and 1, where 0 means any overlap and 1 means complete overlap. Returns ------- pl.DataFrame Original hit data with an additional 'is_primary' column (1 for primary hits, 0 otherwise). All original columns are preserved, with the new column added at the end. Raises ------ KeyError If required columns (chr/peak_id, start, end, hit_similarity) are missing. Notes ----- The algorithm transforms coordinates by scaling by 2 and adjusting by the overlap fraction to create effective overlap regions for efficient processing. This allows using a sweep line algorithm to identify overlaps in a single pass. The transformation works as follows: - Original coordinates: [start, end] - Length = end - start - Adjusted start = start * 2 + length * overlap_frac - Adjusted end = end * 2 - length * overlap_frac This creates regions that overlap only when the original regions have sufficient overlap according to the specified fraction. Examples -------- >>> hits_collapsed = collapse_hits(hits_df, overlap_frac=0.2) >>> primary_hits = hits_collapsed.filter(pl.col("is_primary") == 1) >>> print(f"Kept {primary_hits.height}/{hits_df.height} hits as primary") chrT)maintain_orderci|]\}}|| Sr8)rr*chroms r z!collapse_hits..sBBBHAuuaBBBr ) return_dtypestartendhit_similarity)chrom_id start_trimend_trim similaritypeak_idr@F) allow_copyrArBrCr) is_primary) isinstancepl LazyFramecollectuniqueis_empty enumerateselectcolreplace_strictUInt32castInt32rechunkto_numpyr1 with_columnsSeries) r2r3chroms chrom_to_iddfrrrrrFdf_outs r collapse_hitsr\fsh'2<(($//## U^ " "$ " 7 7F ??   BB &0A0ABBB ^^VE]]11+BI1VVvg*u w/<?EEbhOOPVE]]Q&u w/<?EEbhOOPv.//   ^^VI&&vg*u w/<?EEbhOOPVE]]Q&u w/<?EEbhOOPv.//    B:''5'99I   & &% & 8 8F j> " "e " 4 4Dl#,,,>>L 64FFJ ! !RYz-S-S-S ! T TF Mr hits_dfsrelaxedc |rgd}ngd}t|dkrtdg}|D]Y}t|tjr(||D||Z|d}tdt|D]&}||||dd|dd }'|S) a Intersect hit datasets across multiple runs to find common hits. This function finds hits that appear consistently across multiple Fi-NeMo runs, which can be useful for identifying robust motif instances that are not sensitive to parameter variations or random initialization. Parameters ---------- hits_dfs : List[Union[pl.DataFrame, pl.LazyFrame]] List of hit DataFrames from different Fi-NeMo runs. Each DataFrame must contain the columns specified by the intersection criteria. LazyFrames will be collected before processing. relaxed : bool If True, uses relaxed intersection criteria with only motif names and untrimmed coordinates. If False, uses strict criteria including all coordinate and metadata columns. Returns ------- pl.DataFrame DataFrame containing hits that appear in all input datasets. Columns from later datasets are suffixed with their index (e.g., '_1', '_2'). The first dataset's columns retain their original names. Raises ------ ValueError If fewer than one hits DataFrame is provided. KeyError If required columns for the specified intersection criteria are missing from any of the input DataFrames. Notes ----- Relaxed intersection is useful when comparing results across different region definitions or motif trimming parameters, but may produce less precise matches. Strict intersection requires identical region definitions and is recommended for most use cases. The intersection columns used are: - Relaxed: ["chr", "start_untrimmed", "end_untrimmed", "motif_name", "strand"] - Strict: ["chr", "start", "end", "start_untrimmed", "end_untrimmed", "motif_name", "strand", "peak_name", "peak_id"] The function performs successive inner joins starting with the first DataFrame, so the final result contains only hits present in all input datasets. Examples -------- >>> common_hits = intersect_hits([hits_df1, hits_df2], relaxed=False) >>> print(f"Found {common_hits.height} hits common to both runs") >>> >>> # Compare relaxed vs strict intersection >>> relaxed_hits = intersect_hits([hits_df1, hits_df2], relaxed=True) >>> strict_hits = intersect_hits([hits_df1, hits_df2], relaxed=False) >>> print(f"Relaxed: {relaxed_hits.height}, Strict: {strict_hits.height}") )r5start_untrimmed end_untrimmed motif_namestrand) r5r<r>r`rarbrc peak_namerDr z$At least one hits dataframe requiredrinnerrT)onhowsuffix join_nullscoalesce) len ValueErrorrGrHrIappendrJr#join)r]r^ join_cols collected_dfsrZr2r*s rintersect_hitsrqsx WWW      8}}q?@@@M%% b", ' ' %   . . . .   $ $ $ $AG 1c-(( ) )  ,, ! q77    Nr )__doc__r$typingrrnumpyrrpolarsrHnumbar numba.typesrrr r jaxtypingr r r1 DataFramerIfloatr\boolrqr8r rr|sO   555555555555  F111I fat,,, eQd+++ eQd+++ gq#---  C7D=!C  C gtm C & C  $ CCCCL] 2<- .]>C]\]]]]@d5r|345d@Dd\ddddddr