from typing import Dict, List, Tuple, Optional, Any import os, json, math, numpy as np, pandas as pd from collections import defaultdict, Counter # ---------------- Genetic code (DNA) ---------------- AA2CODONS = { 'A':['GCT','GCC','GCA','GCG'], 'R':['CGT','CGC','CGA','CGG','AGA','AGG'], 'N':['AAT','AAC'], 'D':['GAT','GAC'], 'C':['TGT','TGC'], 'Q':['CAA','CAG'], 'E':['GAA','GAG'], 'G':['GGT','GGC','GGA','GGG'], 'H':['CAT','CAC'], 'I':['ATT','ATC','ATA'], 'L':['TTA','TTG','CTT','CTC','CTA','CTG'], 'K':['AAA','AAG'], 'M':['ATG'], 'F':['TTT','TTC'], 'P':['CCT','CCC','CCA','CCG'], 'S':['TCT','TCC','TCA','TCG','AGT','AGC'], 'T':['ACT','ACC','ACA','ACG'], 'W':['TGG'], 'Y':['TAT','TAC'], 'V':['GTT','GTC','GTA','GTG'], '*':['TAA','TAG','TGA'] } DNA_Codons = { # 'M' - START, '_' - STOP "GCT": "A", "GCC": "A", "GCA": "A", "GCG": "A", "TGT": "C", "TGC": "C", "GAT": "D", "GAC": "D", "GAA": "E", "GAG": "E", "TTT": "F", "TTC": "F", "GGT": "G", "GGC": "G", "GGA": "G", "GGG": "G", "CAT": "H", "CAC": "H", "ATA": "I", "ATT": "I", "ATC": "I", "AAA": "K", "AAG": "K", "TTA": "L", "TTG": "L", "CTT": "L", "CTC": "L", "CTA": "L", "CTG": "L", "ATG": "M", "AAT": "N", "AAC": "N", "CCT": "P", "CCC": "P", "CCA": "P", "CCG": "P", "CAA": "Q", "CAG": "Q", "CGT": "R", "CGC": "R", "CGA": "R", "CGG": "R", "AGA": "R", "AGG": "R", "TCT": "S", "TCC": "S", "TCA": "S", "TCG": "S", "AGT": "S", "AGC": "S", "ACT": "T", "ACC": "T", "ACA": "T", "ACG": "T", "GTT": "V", "GTC": "V", "GTA": "V", "GTG": "V", "TGG": "W", "TAT": "Y", "TAC": "Y", "TAA": "_", "TAG": "_", "TGA": "_" } # ---------------- Helpers ---------------- def aminoacid_percentage(codons): """% and count of codons per amino acid for a chosen codon list.""" amino_dict_count = defaultdict(list) amino_dict_per = defaultdict(list) for v in codons: amino = DNA_Codons[v] amino_dict_per[amino].append(v) for k,v in amino_dict_per.items(): c = Counter(v) sub_dict = {kk:np.round(vv/len(v),2) for kk,vv in c.items()} amino_dict_per[k] = sub_dict amino_dict_count[k] = c return amino_dict_per,amino_dict_count def gc_content(sequence=None,fasta_dir=None): """ GC% overall + per codon position for each sequence provided. Use with: gc_content([nt_string]) """ if sequence is not None: sequences=sequence else: raise ValueError("Provide sequence=[nt_string].") gc_content=[[(seq.lower().count('g')+seq.lower().count('c'))/len(seq)*100] for seq in sequences] for index,seq in enumerate(sequences): seq=seq.lower() for i in range(3): position_nucleotides=seq[i::3] gc_count = position_nucleotides.count('g') + position_nucleotides.count('c') total_position_nucleotides = len(position_nucleotides) gc_content_percentage = (gc_count / max(1,total_position_nucleotides)) * 100 gc_content[index].append(gc_content_percentage) return pd.DataFrame(gc_content,columns=['Original','Position One','Position Two','Position Three']).round(1) def c_content(sequences: List[str]) -> pd.DataFrame: """ C% overall + per codon position for each nucleotide sequence provided. """ c_vals = [[(s.lower().count('c') / max(1, len(s))) * 100] for s in sequences] # <-- fixed extra ')' for idx, seq in enumerate(sequences): seq = seq.lower() for i in range(3): pos_nt = seq[i::3] c_count = pos_nt.count('c') pct = (c_count / max(1, len(pos_nt))) * 100 c_vals[idx].append(pct) return pd.DataFrame( c_vals, columns=['Original C%', 'Position One C%', 'Position Two C%', 'Position Three C%'] ).round(1) def parse_kmer_list(x) -> List[str]: """Parse semicolon-separated kmers 'AAA;TTT;...' into a list (uppercased).""" if x is None: return [] s = str(x).strip() if not s: return [] return [k.strip().upper() for k in s.split(";") if k.strip()] def load_summary(summary_path: str) -> pd.DataFrame: """Read CSV or XLSX summary to DataFrame.""" sp = summary_path.lower() if sp.endswith(".xlsx") or sp.endswith(".xls"): return pd.read_excel(summary_path) return pd.read_csv(summary_path) def scale_interval_codon(a_codon: int, b_codon: int, L_train_cds: int, L_target_cds: int) -> Tuple[int, int]: """Percentage-map [a,b] (1-based codons) from training length to target length.""" a2 = 1 + int(((a_codon - 1) / max(1, L_train_cds)) * max(1, L_target_cds - 1)) b2 = int(math.ceil((b_codon / max(1, L_train_cds)) * L_target_cds)) a2 = max(1, min(a2, L_target_cds)) b2 = max(1, min(b2, L_target_cds)) return a2, b2 def codon_region_to_nt_span(a_codon: int, b_codon: int) -> Tuple[int, int]: """Convert a codon region to 1-based nucleotide span [nt_start, nt_end].""" return 3*(a_codon-1)+1, 3*b_codon def feasible_codons_with_pattern(aa: str, pattern: str) -> List[str]: """ pattern like '.C.' where '.' = free nt, returns codons for this AA matching it. """ outs = [] for c in AA2CODONS[aa]: ok = True for i, ch in enumerate(pattern): if ch != '.' and c[i] != ch: ok = False break if ok: outs.append(c) return outs # ---------------- Wobble preference helpers ---------------- def wobble_bonus(base: str) -> int: """3rd-base preference: C=+2, G=0, A/T=-1; '.' or others -> 0.""" if base == "C": return 2 if base == "G": return 0 if base in ("A", "T"): return -1 return 0 def expected_wobble_for_pattern(aa: str, pattern: str) -> float: """Average wobble bonus if wobble free, or fixed wobble bonus if specified.""" wob = pattern[2] if wob in "ACGT": return float(wobble_bonus(wob)) feas = feasible_codons_with_pattern(aa, pattern) if not feas: return 0.0 return sum(wobble_bonus(c[2]) for c in feas) / len(feas) def placement_wobble_score(constraints: Dict[int, str], aa_seq: str) -> float: """Sum wobble preferences across codons touched by a seed placement.""" total = 0.0 for ci, patt in constraints.items(): if 0 <= ci < len(aa_seq): total += expected_wobble_for_pattern(aa_seq[ci], patt) return total # ---------------- Seeding best_kmers (enumerate placements) ---------------- def place_kmer_seed_in_region_codon(aa_seq: str, a_codon: int, b_codon: int, kmer: str, fixed_nt: Dict[int, str]) -> List[Dict]: """ Enumerate feasible placements of a k-mer inside [a_codon,b_codon] (1-based, inclusive). Returns list of dicts with {"start_nt","end_nt","constraints"} where nt indices are 0-based. """ nt_start, nt_end = codon_region_to_nt_span(a_codon, b_codon) region_start_nt0, region_end_nt0 = nt_start-1, nt_end-1 k = len(kmer) placements = [] for u in range(region_start_nt0, region_end_nt0 - k + 2): v = u + k - 1 # conflict with fixed nts? if any((u+t) in fixed_nt and fixed_nt[u+t] != kmer[t] for t in range(k)): continue # build codon-level constraints constraints: Dict[int, str] = {} codon_i0 = u // 3 codon_i1 = v // 3 ok = True for ci in range(codon_i0, codon_i1 + 1): if ci >= len(aa_seq): ok = False; break patt = list("...") for ofs in range(3): nt_idx = ci*3 + ofs if u <= nt_idx <= v: patt[ofs] = kmer[nt_idx - u] patt_s = "".join(patt) if not feasible_codons_with_pattern(aa_seq[ci], patt_s): ok = False; break constraints[ci] = patt_s if not ok: continue placements.append({"start_nt": u, "end_nt": v, "constraints": constraints}) return placements # ---------------- Scoring while filling ---------------- def _parse_klist(s): if pd.isna(s) or not str(s).strip(): return [] return [t.strip().upper() for t in str(s).split(";") if t.strip()] def _parse_kwmap(s): """Parse 'kmer:weight;kmer:weight;...' -> dict. If weights absent, return {}.""" if pd.isna(s) or not str(s).strip(): return {} out = {} for tok in str(s).split(";"): tok = tok.strip() if not tok: continue if ":" in tok: k, w = tok.split(":", 1) try: out[k.strip().upper()] = float(w) except: pass return out def len_weight(K: int, is_pos: bool) -> float: """ Length-aware default weights. Pos: +0.5*K; Neg: -1.0*K """ return (0.5 * K) if is_pos else (-1.0 * K) def collect_right_known_nt(ci: int, fixed_nt: dict, L_target_cds: int, limit_nt: int) -> str: """ Collect up to 'limit_nt' contiguous known nts to the RIGHT of codon index 'ci'. """ out = [] start_nt = (ci + 1) * 3 nt = start_nt while len(out) < limit_nt: if nt in fixed_nt: out.append(fixed_nt[nt]) nt += 1 continue break return "".join(out) def enumerate_local_windows_overlapping_new(block: str, new_start: int, # index in block (0-based) new_len: int, K: int): """Yield all K-length windows inside 'block' that overlap [new_start, new_start+new_len).""" L = len(block) if K > L: return new_end = new_start + new_len # exclusive s_min = max(0, new_end - K) s_max = min(new_start, L - K) s_lo = max(0, new_start - (K - 1)) s_hi = min(L - K, new_end - 1) s_from = min(s_min, s_lo) s_to = max(s_max, s_hi) for s in range(s_from, s_to + 1): if s < new_end and (s + K) > new_start: yield s, block[s:s+K] def score_increment_multiKs(tail_nt: str, new_codon: str, best_sets: dict, # {K: set(kmer)} (we'll pass {} during fill) avoid_sets: dict, # {K: set(kmer)} pos_w: dict = None, # {(K,kmer): weight} neg_w: dict = None, # {(K,kmer): weight} wobble: bool = True, wobble_scale: float = 1.0, scoring_mode: str = "local", # "local" or "suffix" right_known_nt: str = "", hard_forbid: set = None, require_full_windows: bool = True): """ Incremental score for appending `new_codon`. - Only FULL K-mer windows that are fully known are scored. - We build a local block = left_tail + new_codon + right_known_nt (all known nts). """ HARD_KILL = -1e9 # Build local block if scoring_mode == "local": left = tail_nt or "" mid = new_codon right = right_known_nt or "" block = left + mid + right new_start = len(left) new_len = 3 L_block = len(block) else: block = (tail_nt + new_codon) if tail_nt else new_codon new_start = len(block) - len(new_codon) new_len = 3 L_block = len(block) gain = 0.0 Ks = sorted(best_sets.keys() | avoid_sets.keys()) for K in Ks: if K > L_block: continue for _, km in enumerate_local_windows_overlapping_new(block, new_start, new_len, K): if len(km) != K: continue if hard_forbid and km in hard_forbid: return HARD_KILL # negatives (penalty) if km in avoid_sets.get(K, set()): if neg_w and (K, km) in neg_w: gain += -abs(neg_w[(K, km)]) else: gain += len_weight(K, is_pos=False) # positives (reward) — left as option; we pass {} during fill if km in best_sets.get(K, set()): if pos_w and (K, km) in pos_w: gain += abs(pos_w[(K, km)]) else: gain += len_weight(K, is_pos=True) if wobble: gain += wobble_scale * wobble_bonus(new_codon[2]) return gain # ---------------- Row parsing helpers ---------------- def _parse_allowed_Ks_from_row(row) -> List[int] | None: s = row.get("K_allowed", None) if isinstance(s, str) and s.strip(): out = [] for tok in s.split(";"): tok = tok.strip() if tok: try: out.append(int(tok)) except: pass return out or None return None def _row_has_best(row) -> bool: """True if this region has any best_kmers (positives) to seed.""" return bool(_parse_klist(row.get("best_kmers", ""))) def _row_has_avoid(row) -> bool: """True if this region provides any avoid motifs.""" if _parse_klist(row.get("avoid_kmers", "")): return True for K in range(2, 10): if _parse_klist(row.get(f"K{K}_neg", "")): return True if str(row.get(f"K{K}_neg_w", "")).strip(): return True if str(row.get(f"K{K}_neg_norm", "")).strip(): return True return False def _build_avoid_sets_from_row(row, allowed_Ks=None): """ Return NEGATIVE scoring sets + weights: avoid_sets: {K: set(kmer)} neg_w: {(K,kmer): weight} """ avoid_sets, neg_w = {}, {} # From avoid_kmers (variable K) for km in parse_kmer_list(row.get("avoid_kmers", "")): K = len(km) if allowed_Ks is not None and K not in allowed_Ks: continue avoid_sets.setdefault(K, set()).add(km.upper()) # From per-K columns for K in range(2, 10): if allowed_Ks is not None and K not in allowed_Ks: continue neg_col = f"K{K}_neg" if neg_col in row and pd.notna(row[neg_col]): for km in _parse_klist(row.get(neg_col, "")): if len(km) == K: avoid_sets.setdefault(K, set()).add(km.upper()) negw_col = f"K{K}_neg_w" if negw_col in row and pd.notna(row[negw_col]): for km, w in _parse_kwmap(row.get(negw_col, "")).items(): if len(km) == K: neg_w[(K, km.upper())] = float(w) return avoid_sets, neg_w def _compute_target_intervals(df: pd.DataFrame, L_train_cds: int, L_target_cds: int, use_percent_intervals: bool = True): """ Use explicit 'tgt_start','tgt_end' if present; else scale if use_percent_intervals=True; else clip original [start,end] to target length. Returns list of tuples: (tgt_a, tgt_b, row). """ has_tgt = ("tgt_start" in df.columns and "tgt_end" in df.columns) out = [] for _, r in df.iterrows(): a_c, b_c = int(r["start"]), int(r["end"]) if has_tgt and pd.notna(r.get("tgt_start")) and pd.notna(r.get("tgt_end")): ta, tb = int(r["tgt_start"]), int(r["tgt_end"]) elif use_percent_intervals: ta, tb = scale_interval_codon(a_c, b_c, L_train_cds, L_target_cds) else: ta, tb = max(1, a_c), min(L_target_cds, b_c) if ta <= tb: out.append((ta, tb, r)) return out # ---------------- Seeding (positives) ---------------- def build_promote_queue(row, allowed_Ks: List[int] | None = None, Ks_in_order: List[int] | None = None, exclude_winner_K_from_extras: bool = True) -> List[str]: """ Build a seeding queue. (We won't use this if only_best=True; kept here for completeness.) """ base = [km.upper() for km in _parse_klist(row.get("best_kmers",""))] if allowed_Ks: S = set(allowed_Ks) base = [km for km in base if len(km) in S] winner = base[0] if base else None queue = base[:] K_order = (Ks_in_order or allowed_Ks or list(range(2,11))).copy() if exclude_winner_K_from_extras and winner: K_order = [K for K in K_order if K != len(winner)] seen_in_queue = set(queue) extras_by_K: dict[int, dict[str, float]] = {K: {} for K in K_order} for K in K_order: for col in (f"K{K}_pos_w", f"K{K}_pos_norm", f"K{K}_pos"): cell = row.get(col, "") if not isinstance(cell, str) or not cell.strip(): continue if col.endswith(("_pos_w","_pos_norm")): for km, sc in _parse_kwmap(cell).items(): km = km.upper() if len(km) != K or km in seen_in_queue: continue prev = extras_by_K[K].get(km, float("-inf")) scf = float(sc) if scf > prev: extras_by_K[K][km] = scf else: for km in _parse_klist(cell): km = km.upper() if len(km) != K or km in seen_in_queue: continue extras_by_K[K].setdefault(km, 0.0) for K in K_order: items = sorted(extras_by_K[K].items(), key=lambda kv: (-kv[1], kv[0])) queue.extend([km for km, _ in items]) out, seen = [], set() for km in queue: if km not in seen: out.append(km); seen.add(km) return out def seed_promote_motifs(aa_seq: str, a_c: int, b_c: int, row: pd.Series, fixed_nt: Dict[int, str], use_wobble: bool = True, wobble_scale: float = 1.0, seed_cap: int = 3, min_gap_codons: int = 2, allowed_Ks: List[int] | None = None, only_best: bool = False) -> tuple[List[str], List[str]]: """ Greedy seeding of Promote motifs (positives). If only_best=True, seed strictly from 'best_kmers' and ignore Kx_pos* extras. """ if only_best: queue = [km.upper() for km in _parse_klist(row.get("best_kmers", ""))] if allowed_Ks: S = set(allowed_Ks) queue = [km for km in queue if len(km) in S] else: queue = build_promote_queue(row, allowed_Ks=allowed_Ks) if not queue: return [], [] placed, skipped = [], [] placed_starts_nt: List[int] = [] min_gap_nt = 3 * max(1, int(min_gap_codons)) for km in queue: if seed_cap is not None and len(placed) >= seed_cap: break places = place_kmer_seed_in_region_codon(aa_seq, a_c, b_c, km, fixed_nt) if not places: skipped.append(km); continue def _ok_start_gap(p): s_nt = p["start_nt"] return all(abs(s_nt - ps) >= min_gap_nt for ps in placed_starts_nt) feas = [p for p in places if _ok_start_gap(p)] if not feas: skipped.append(km); continue best_p, best_s = None, float("-inf") for p in feas: s = 0.0 if use_wobble: s += wobble_scale * placement_wobble_score(p["constraints"], aa_seq) s += -0.001 * p["start_nt"] if s > best_s: best_s, best_p = s, p if best_p is None: skipped.append(km); continue u, v = best_p["start_nt"], best_p["end_nt"] if any((t in fixed_nt and fixed_nt[t] != km[t - u]) for t in range(u, v + 1)): skipped.append(km); continue for t in range(u, v + 1): fixed_nt[t] = km[t - u] placed_starts_nt.append(u) placed.append(km) return placed, skipped # ---------------- Main: seed positives; avoid-only scoring on fill ---------------- def optimization(summary_path: str,aa_seq: str, use_wobble: bool = True, wobble_scale: float = 1.0, use_percent_intervals: bool = True, seed_cap: int | None = 3, min_gap_codons: int = 2, beam_size: int = 5): """ - Seed ONLY rows that have `best_kmers` (positives). - Global left->right beam fill. - Inside mapped regions that have avoiders, apply NEGATIVE scoring (avoid sets). No positive rewards during fill. Outside all regions: wobble only. Returns: (designed_nt, aa_stats_stub, gc_percent_df, log_info) """ df = load_summary(summary_path) L_target_cds = len(aa_seq) L_train_cds = int(df["end"].max()) # 1) map every summary row to a target interval intervals = _compute_target_intervals(df, L_train_cds, L_target_cds, use_percent_intervals) # 2) seed positives (best_kmers) only fixed_nt: Dict[int, str] = {} seed_log: Dict[Tuple[int,int], Dict[str, Any]] = {} for (ta, tb, row) in intervals: if not _row_has_best(row): continue allowed_Ks = _parse_allowed_Ks_from_row(row) placed, skipped = seed_promote_motifs( aa_seq=aa_seq, a_c=ta, b_c=tb, row=row, fixed_nt=fixed_nt, use_wobble=use_wobble, wobble_scale=wobble_scale, seed_cap=seed_cap, min_gap_codons=min_gap_codons, allowed_Ks=allowed_Ks, only_best=True ) seed_log[(ta, tb)] = {"placed": placed, "skipped": skipped} # 3) build per-codon avoid-only config per_ci_cfg: List[Optional[Tuple[Dict[int,set], Dict[Tuple[int,str],float], int]]] = [None] * L_target_cds Kmax_all = 2 for (ta, tb, row) in intervals: if not _row_has_avoid(row): continue allowed_Ks = _parse_allowed_Ks_from_row(row) avoid_sets, neg_w = _build_avoid_sets_from_row(row, allowed_Ks) if not avoid_sets: continue Kmax_here = max(avoid_sets.keys()) Kmax_all = max(Kmax_all, Kmax_here) for ci in range(ta - 1, tb): if per_ci_cfg[ci] is None: per_ci_cfg[ci] = ({K:set(v) for K,v in avoid_sets.items()}, dict(neg_w), Kmax_here) else: old_avoid, old_negw, old_kmax = per_ci_cfg[ci] for K, ss in avoid_sets.items(): old_avoid.setdefault(K, set()).update(ss) old_negw.update(neg_w) per_ci_cfg[ci] = (old_avoid, old_negw, max(old_kmax, Kmax_here)) # 4) one global left->right beam fill (avoid-only scoring, wobble everywhere) k_len_ref = Kmax_all tail_nt = "" beam: List[Tuple[float, str, List[str], Dict[int,str]]] = [(0.0, tail_nt, [], dict())] for ci in range(L_target_cds): aa = aa_seq[ci] # enforce seeded nts patt = list("...") for ofs in range(3): nt_idx = ci * 3 + ofs if nt_idx in fixed_nt: patt[ofs] = fixed_nt[nt_idx] patt_s = "".join(patt) cand_codons = feasible_codons_with_pattern(aa, patt_s) or AA2CODONS[aa] cfg = per_ci_cfg[ci] if cfg is None: avoid_sets, neg_w, Kmax_here = {}, {}, 2 else: avoid_sets, neg_w, Kmax_here = cfg new_beam = [] for score, tail, local_codons, local_fix in beam: right_known = collect_right_known_nt(ci, fixed_nt, L_target_cds, limit_nt=Kmax_here-1) for c in cand_codons: gain = score_increment_multiKs( tail_nt=tail, new_codon=c, best_sets={}, # << NO positive rewards during fill avoid_sets=avoid_sets, pos_w=None, neg_w=neg_w if neg_w else None, wobble=use_wobble, wobble_scale=wobble_scale, scoring_mode="local", right_known_nt=right_known, hard_forbid=None ) if gain <= -1e5: continue tail2 = (tail + c)[-(k_len_ref - 1):] if k_len_ref > 1 else "" local2 = local_codons + [c] fix2 = dict(local_fix) for ofs, ch in enumerate(c): fix2[ci * 3 + ofs] = ch new_beam.append((score + gain, tail2, local2, fix2)) if not new_beam: # safety fallback c = cand_codons[0] tail2 = (beam[0][1] + c)[-(k_len_ref - 1):] if k_len_ref > 1 else "" local2 = beam[0][2] + [c] fix2 = dict(beam[0][3]) for ofs, ch in enumerate(c): fix2[ci * 3 + ofs] = ch new_beam = [(beam[0][0], tail2, local2, fix2)] new_beam.sort(key=lambda t: t[0], reverse=True) beam = new_beam[:beam_size] best_score, _, best_local, best_fix = max(beam, key=lambda t: t[0]) # 5) finalize chosen_codons = best_local fixed_nt.update(best_fix) designed_nt = "".join(chosen_codons) gc_percent = gc_content([designed_nt]) c_percent = c_content([designed_nt]) # Build amino-acid → {codon: fraction} dict for UI aa_percent_dict, _aa_counts = aminoacid_percentage(chosen_codons) log_info = [{ "mode": "global_fill_avoid_only", "beam_best_score": best_score, "beam_kept": beam_size, "seed_summary": seed_log }] # aa_percent placeholder kept for API compatibility return designed_nt, aa_percent_dict, gc_percent, c_percent, log_info # ---------------- Example usage ---------------- if __name__ == "__main__": # Example: # summary_path = "region_sweep_summary.csv" # aa_seq = "M" + "A"*514 # 515 aa example # nt_seq, aa_stats, gc_df, log = optimization( # summary_path=summary_path, # aa_seq=aa_seq, # use_wobble=True, # wobble_scale=1.0, # use_percent_intervals=False, # True to scale; False if you have tgt_start/tgt_end # seed_cap=3, # min_gap_codons=2, # beam_size=5 # ) # print(nt_seq) # print(gc_df) pass