|
| 1 | +from geneml.types import SplicingType, Transcript |
| 2 | + |
| 3 | + |
| 4 | +def get_ordered_introns(exons: list, strand: int) -> list[tuple[int,int]]: |
| 5 | + """ |
| 6 | + Returns introns (donor, acceptor) in transcriptional order. |
| 7 | +
|
| 8 | + Args: |
| 9 | + exons: list of Exon objects, sorted by genomic position. |
| 10 | + strand: +1 for forward strand, -1 for reverse strand. |
| 11 | +
|
| 12 | + Returns: |
| 13 | + List of introns as (donor, acceptor) tuples in transcriptional order. |
| 14 | + """ |
| 15 | + if len(exons) == 1: |
| 16 | + return [] # single-exon transcript has no introns |
| 17 | + |
| 18 | + introns = [] |
| 19 | + for i in range(len(exons) - 1): |
| 20 | + # donor = 5' splice site, acceptor = 3' splice site (transcriptional) |
| 21 | + if strand == 1: |
| 22 | + donor = exons[i].end |
| 23 | + acceptor = exons[i + 1].start |
| 24 | + elif strand == -1: |
| 25 | + donor = exons[i].start |
| 26 | + acceptor = exons[i + 1].end |
| 27 | + else: |
| 28 | + raise ValueError(f"Invalid strand: {strand}") |
| 29 | + introns.append((donor, acceptor)) |
| 30 | + return introns |
| 31 | + |
| 32 | + |
| 33 | +def get_introns_in_range(introns: list[tuple[int,int]], range: tuple[int,int], strand: int |
| 34 | + ) -> list[tuple[int,int]]: |
| 35 | + """Returns introns that fall within the specified genomic range. |
| 36 | +
|
| 37 | + Args: |
| 38 | + introns: List of intron tuples (donor, acceptor) in transcriptional order. |
| 39 | + range: Tuple of (start, end) genomic coordinates defining the range. |
| 40 | + strand: +1 for forward strand, -1 for reverse strand. |
| 41 | +
|
| 42 | + Returns: |
| 43 | + List of introns that fall within the specified range. |
| 44 | + """ |
| 45 | + if strand == 1: |
| 46 | + return [(s, e) for s, e in introns if s >= range[0] and e <= range[1]] |
| 47 | + elif strand == -1: |
| 48 | + return [(s, e) for s, e in introns if s <= range[1] and e >= range[0]] |
| 49 | + else: |
| 50 | + raise ValueError(f"Invalid strand: {strand}") |
| 51 | + |
| 52 | + |
| 53 | +def get_alternative_splicing_type(primary: Transcript, alt: Transcript) -> SplicingType: |
| 54 | + """Classify alternative splicing events between a primary and alternative transcript. |
| 55 | +
|
| 56 | + Compares intron junctions and terminal exon boundaries to detect exon skipping, |
| 57 | + alternative 3' and 5' splice sites, intron retention, and alternative first/last |
| 58 | + exons. If multiple event types are detected, the transcript is labeled as |
| 59 | + complex. |
| 60 | +
|
| 61 | + Args: |
| 62 | + primary: The reference transcript to compare against. |
| 63 | + alt: The alternative transcript being classified. |
| 64 | +
|
| 65 | + Returns: |
| 66 | + The assigned SplicingType for the alternative transcript. |
| 67 | + """ |
| 68 | + assert primary is not alt, "Should not compare to self" |
| 69 | + |
| 70 | + events = set() |
| 71 | + strand = primary.strand |
| 72 | + |
| 73 | + # Order by transcriptional order |
| 74 | + P = get_ordered_introns(primary.exons, strand) |
| 75 | + A = get_ordered_introns(alt.exons, strand) |
| 76 | + |
| 77 | + # Only compare introns within the shared genomic region of the two transcripts |
| 78 | + shared_range = (max(primary.start, alt.start), min(primary.end, alt.end)) |
| 79 | + P = get_introns_in_range(P, shared_range, strand) |
| 80 | + A = get_introns_in_range(A, shared_range, strand) |
| 81 | + |
| 82 | + P_set = set(P) |
| 83 | + A_set = set(A) |
| 84 | + |
| 85 | + # Track junctions consumed by exon skipping to avoid double-counting them |
| 86 | + consumed_P = set() |
| 87 | + consumed_A = set() |
| 88 | + |
| 89 | + # 1. ALTERNATIVE FIRST / LAST EXON |
| 90 | + if strand == 1: |
| 91 | + if alt.exons[0].start != primary.exons[0].start: |
| 92 | + events.add(SplicingType.ALT_FIRST_EXON) |
| 93 | + if alt.exons[-1].end != primary.exons[-1].end: |
| 94 | + events.add(SplicingType.ALT_LAST_EXON) |
| 95 | + elif strand == -1: |
| 96 | + if alt.exons[0].end != primary.exons[0].end: |
| 97 | + events.add(SplicingType.ALT_FIRST_EXON) |
| 98 | + if alt.exons[-1].start != primary.exons[-1].start: |
| 99 | + events.add(SplicingType.ALT_LAST_EXON) |
| 100 | + else: |
| 101 | + raise ValueError(f"Invalid strand: {strand}") |
| 102 | + |
| 103 | + # 2. EXON SKIPPING |
| 104 | + for s, e in P: |
| 105 | + for i in range(len(A) - 1): |
| 106 | + s1, e1 = A[i] |
| 107 | + s2, e2 = A[i + 1] |
| 108 | + |
| 109 | + if s1 == s and e2 == e: |
| 110 | + events.add(SplicingType.EXON_SKIPPING) |
| 111 | + consumed_P.add((s, e)) |
| 112 | + consumed_A.add((s1, e1)) |
| 113 | + consumed_A.add((s2, e2)) |
| 114 | + |
| 115 | + for s, e in A: |
| 116 | + for i in range(len(P) - 1): |
| 117 | + s1, e1 = P[i] |
| 118 | + s2, e2 = P[i + 1] |
| 119 | + |
| 120 | + if s1 == s and e2 == e: |
| 121 | + events.add(SplicingType.EXON_SKIPPING) |
| 122 | + consumed_A.add((s, e)) |
| 123 | + consumed_P.add((s1, e1)) |
| 124 | + consumed_P.add((s2, e2)) |
| 125 | + |
| 126 | + # 3. ALTERNATIVE 3' / 5' SPLICE SITES |
| 127 | + for s1, e1 in P: |
| 128 | + if (s1, e1) in consumed_P: |
| 129 | + continue |
| 130 | + for s2, e2 in A: |
| 131 | + if (s2, e2) in consumed_A: |
| 132 | + continue |
| 133 | + if s1 == s2 and e1 != e2: |
| 134 | + # Skip if this is the terminal exon boundary (already counted as ALT_LAST_EXON) |
| 135 | + if (s1, e1) == P[-1] or (s2, e2) == A[-1] and SplicingType.ALT_LAST_EXON in events: |
| 136 | + pass |
| 137 | + else: |
| 138 | + events.add(SplicingType.ALT_3_SPLICE_SITE) |
| 139 | + consumed_P.add((s1, e1)) |
| 140 | + consumed_A.add((s2, e2)) |
| 141 | + if e1 == e2 and s1 != s2: |
| 142 | + events.add(SplicingType.ALT_5_SPLICE_SITE) |
| 143 | + consumed_P.add((s1, e1)) |
| 144 | + consumed_A.add((s2, e2)) |
| 145 | + |
| 146 | + # 4. INTRON RETENTION |
| 147 | + remaining_P = (P_set - A_set) - consumed_P |
| 148 | + remaining_A = (A_set - P_set) - consumed_A |
| 149 | + |
| 150 | + if remaining_P or remaining_A: |
| 151 | + events.add(SplicingType.INTRON_RETENTION) |
| 152 | + |
| 153 | + |
| 154 | + # Assign final splicing type |
| 155 | + if not events: |
| 156 | + return SplicingType.UNKNOWN |
| 157 | + elif len(events) == 1: |
| 158 | + return events.pop() |
| 159 | + return SplicingType.COMPLEX |
0 commit comments