Skills › Scientific Agent Skills › primer-design › _common.py
_common.py
134 lines · skills/primer-design/scripts/_common.py at 5f2bc03
1"""Validated local inputs and provenance shared by primer-design tools."""
2from __future__ import annotations
4import csv
5import hashlib
6import json
7import math
8import os
9import re
10import tempfile
11from pathlib import Path
13IUPAC = "ACGTRYSWKMBDHVN"
14PAIR_FIELDS = ("pair_id", "forward", "reverse", "forward_tail", "reverse_tail")
17def dna(value: str, label: str, *, ambiguous: bool = False, empty: bool = False) -> str:
18 if not isinstance(value, str):
19 raise ValueError(f"{label}: sequence must be a string")
20 seq = value.strip().upper()
21 if (not seq and not empty) or any(c not in (IUPAC if ambiguous else "ACGT") for c in seq):
22 raise ValueError(f"{label}: expected {'IUPAC' if ambiguous else 'ACGT'} DNA, 5-prime to 3-prime, without internal whitespace")
23 return seq
26def revcomp(seq: str) -> str:
27 return seq.translate(str.maketrans(IUPAC, "TGCAYRSWMKVHDBN"))[::-1]
30def read_fasta(path) -> dict[str, str]:
31 records, chunks, current = {}, [], None
32 with Path(path).open(encoding="utf-8-sig") as handle:
33 for number, raw in enumerate(handle, 1):
34 line = raw.strip()
35 if not line:
36 continue
37 if line.startswith(">"):
38 if current is not None:
39 records[current] = dna("".join(chunks), current, ambiguous=True)
40 current = line[1:].split()[0] if line[1:].split() else ""
41 if not current or current in records:
42 raise ValueError(f"FASTA line {number}: empty or duplicate record ID")
43 chunks = []
44 else:
45 if current is None:
46 raise ValueError(f"FASTA line {number}: sequence before header")
47 chunks.append(line)
48 if current is not None:
49 records[current] = dna("".join(chunks), current, ambiguous=True)
50 if not records:
51 raise ValueError("FASTA contains no records")
52 return records
55def read_pairs(path) -> list[dict[str, str]]:
56 with Path(path).open(newline="", encoding="utf-8-sig") as handle:
57 reader = csv.DictReader(handle, delimiter="\t")
58 header = reader.fieldnames or []
59 if len(set(header)) != len(header) or not {"pair_id", "forward", "reverse"} <= set(header):
60 raise ValueError("Pair TSV needs unique columns pair_id, forward, reverse; optional forward_tail, reverse_tail")
61 if set(header) - set(PAIR_FIELDS):
62 raise ValueError(f"Unknown pair TSV columns: {sorted(set(header) - set(PAIR_FIELDS))}")
63 result, seen = [], set()
64 for number, row in enumerate(reader, 2):
65 if None in row or any(v is None for v in row.values()):
66 raise ValueError(f"Pair TSV line {number}: wrong column count")
67 pair_id = row["pair_id"].strip()
68 if not re.fullmatch(r"[A-Za-z0-9_.-]+", pair_id) or pair_id in seen:
69 raise ValueError(f"Pair TSV line {number}: duplicate or invalid pair_id")
70 seen.add(pair_id)
71 result.append({"pair_id": pair_id, **{
72 field: dna(row.get(field, ""), f"{pair_id}.{field}", empty=field.endswith("_tail"))
73 for field in PAIR_FIELDS[1:]
74 }})
75 if not result:
76 raise ValueError("Pair TSV contains no primer pairs")
77 return result
80def sha256_file(path) -> str:
81 digest = hashlib.sha256()
82 with Path(path).open("rb") as handle:
83 for chunk in iter(lambda: handle.read(1024 * 1024), b""):
84 digest.update(chunk)
85 return digest.hexdigest()
88def check_output_paths(inputs, outputs):
89 sources = {Path(p).resolve() for p in inputs if p is not None}
90 destinations = [Path(p).resolve() for p in outputs if p is not None]
91 if len(set(destinations)) != len(destinations) or sources.intersection(destinations):
92 raise ValueError("Output paths must be distinct from each other and all inputs")
93 paths = list(sources) + destinations
94 for index, path in enumerate(paths):
95 if path.exists():
96 for other in paths[index + 1:]:
97 if other.exists() and path.samefile(other):
98 raise ValueError("Output paths must not be hardlink aliases of inputs or other outputs")
101def write_json(path, data):
102 """Replace a complete JSON document atomically, never serialize NaN/Infinity."""
103 destination = Path(path)
104 serialized = json.dumps(data, indent=2, allow_nan=False) + "\n"
105 destination.parent.mkdir(parents=True, exist_ok=True)
106 with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=destination.parent, delete=False) as handle:
107 temporary = Path(handle.name)
108 handle.write(serialized)
109 try:
110 os.replace(temporary, destination)
111 finally:
112 temporary.unlink(missing_ok=True)
115def load_primer3():
116 try:
117 import primer3
118 except ImportError as exc:
119 raise ValueError("primer3-py is not installed; install primer3-py==2.3.1 in an isolated environment") from exc
120 return primer3
123def finite_number(value, name, minimum=0, *, positive=False):
124 if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
125 raise ValueError(f"{name} must be a finite number")
126 if value < minimum or (positive and value == minimum):
127 raise ValueError(f"{name} must be {'greater than' if positive else 'at least'} {minimum}")
128 return value
131def integer(value, name, minimum=0):
132 if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
133 raise ValueError(f"{name} must be an integer >= {minimum}")
134 return value