"""Input parsing with format sniffing and structured, user-facing issues. The readers return the same frames as ``ppi_benchmark_controls.py`` (protein_id, sequence) and (protein_a, protein_b, label), in file order, so downstream algorithms are unchanged. They are more forgiving about layout (delimiter and header are detected from content, not the file extension; labels such as ``1.0``/``yes``/``positive`` are accepted) and report problems as ``InputIssue`` records that the web interface can show next to the offending rows. """ from __future__ import annotations import io as _io import re from dataclasses import asdict, dataclass, field from pathlib import Path import numpy as np import pandas as pd from .constants import AMINO_ACIDS PROTEIN_HEADER_NAMES = {"protein_id", "protein", "id", "uniprot", "accession", "sequence", "seq", "species", "organism", "taxon"} PAIR_HEADER_NAMES = {"protein_a", "protein_b", "label", "id_a", "id_b", "interactor_a", "interactor_b", "a", "b", "protein1", "protein2", "uniprot_a", "uniprot_b", "y", "class", "interaction"} SPECIES_HEADER_NAMES = {"protein_id", "species", "organism", "taxon", "taxon_id", "id"} DELIMITERS = {"\t": "tab", ",": "comma", ";": "semicolon", "|": "pipe"} POSITIVE_LABELS = {"1", "1.0", "true", "t", "yes", "y", "pos", "positive", "+", "interact", "interacting", "interaction"} NEGATIVE_LABELS = {"0", "0.0", "false", "f", "no", "n", "neg", "negative", "-", "non-interacting", "noninteracting", "none"} SEQUENCE_CHARS = set("ABCDEFGHIJKLMNOPQRSTUVWXYZ*-") @dataclass class InputIssue: """A problem found in an input file. ``severity`` is ``error`` (blocks the run), ``warning`` (handled automatically, reported) or ``info``.""" code: str severity: str message: str file: str = "" rows: list = field(default_factory=list) # 1-based line numbers (header counts as line 1) examples: list = field(default_factory=list) hint: str = "" def to_dict(self) -> dict: return asdict(self) class InputError(ValueError): """Raised when inputs cannot be used. Carries the full issue list.""" def __init__(self, issues: list[InputIssue]): self.issues = issues errors = [i for i in issues if i.severity == "error"] super().__init__("; ".join(i.message for i in errors) or "invalid input") @dataclass class TableSniff: kind: str # "fasta" or "table" delimiter: str | None delimiter_name: str has_header: bool n_rows: int n_columns: int columns: list[str] preview: list[list[str]] def to_dict(self) -> dict: return asdict(self) # --------------------------------------------------------------------------- # low-level helpers # --------------------------------------------------------------------------- def _read_text(source) -> str: if isinstance(source, (bytes, bytearray)): text = bytes(source).decode("utf-8-sig", errors="replace") elif isinstance(source, Path) or (isinstance(source, str) and "\n" not in source and len(source) < 4096 and Path(source).is_file()): text = Path(source).read_bytes().decode("utf-8-sig", errors="replace") else: text = str(source) return text.replace("\r\n", "\n").replace("\r", "\n") def _content_lines(text: str) -> list[str]: return [ln for ln in text.split("\n") if ln.strip() and not ln.lstrip().startswith("#")] def detect_delimiter(lines: list[str]) -> str | None: """Delimiter whose per-line count is >= 1 and most consistent over the first lines. Returns None for whitespace-separated tables.""" sample = lines[:50] best, best_score = None, 0.0 for d in DELIMITERS: counts = [ln.count(d) for ln in sample] if not counts or min(counts) == 0: continue mode = max(set(counts), key=counts.count) score = counts.count(mode) / len(counts) + (0.01 if d == "\t" else 0.0) if score > best_score: best, best_score = d, score return best def _split_table(text: str) -> tuple[list[list[str]], str | None, list[int]]: """Rows of cells, the delimiter, and the 1-based source line number of each row.""" raw_lines = text.split("\n") kept = [(i + 1, ln) for i, ln in enumerate(raw_lines) if ln.strip() and not ln.lstrip().startswith("#")] delim = detect_delimiter([ln for _, ln in kept]) rows = [] for _, ln in kept: cells = ln.split(delim) if delim else re.split(r"\s+", ln.strip()) rows.append([c.strip().strip('"').strip("'") for c in cells]) return rows, delim, [n for n, _ in kept] def is_fasta(text: str) -> bool: for ln in text.split("\n"): if ln.strip(): return ln.lstrip().startswith(">") return False def _looks_like_sequence(cell: str) -> bool: return len(cell) >= 5 and set(cell.upper()) <= SEQUENCE_CHARS def _header_by_names(row: list[str], names: set[str]) -> bool: return any(c.strip().lower() in names for c in row) def coerce_label(value: str): v = str(value).strip().lower() if v in POSITIVE_LABELS: return 1 if v in NEGATIVE_LABELS: return 0 return None # --------------------------------------------------------------------------- # sniffing (used by the upload preview and the format checker) # --------------------------------------------------------------------------- def sniff(source, kind: str = "pairs", max_preview: int = 8) -> TableSniff: """Describe an input file without fully parsing it. ``kind`` is proteins|pairs|species.""" text = _read_text(source) if is_fasta(text): records = [ln[1:].split()[0] for ln in text.split("\n") if ln.startswith(">") and ln[1:].split()] return TableSniff("fasta", None, "fasta", False, len(records), 2, ["protein_id", "sequence"], [[r, "…"] for r in records[:max_preview]]) rows, delim, _ = _split_table(text) names = {"proteins": PROTEIN_HEADER_NAMES, "pairs": PAIR_HEADER_NAMES, "species": SPECIES_HEADER_NAMES}[kind] has_header = bool(rows) and _header_by_names(rows[0], names) if rows and not has_header and kind == "proteins" and len(rows[0]) >= 2 and len(rows) > 1: has_header = not _looks_like_sequence(rows[0][1]) and _looks_like_sequence(rows[1][1]) if rows and not has_header and kind == "pairs" and len(rows[0]) >= 3 and len(rows) > 1: has_header = coerce_label(rows[0][2]) is None and coerce_label(rows[1][2]) is not None n_cols = max((len(r) for r in rows[:50]), default=0) if has_header: columns = [c or f"column_{i + 1}" for i, c in enumerate(rows[0])] else: default = {"proteins": ["protein_id", "sequence", "species"], "pairs": ["protein_a", "protein_b", "label"], "species": ["protein_id", "species"]}[kind] columns = [default[i] if i < len(default) else f"column_{i + 1}" for i in range(n_cols)] body = rows[1:] if has_header else rows preview = [[(c[:40] + "…") if len(c) > 41 else c for c in r] for r in body[:max_preview]] return TableSniff("table", delim, DELIMITERS.get(delim, "whitespace"), has_header, len(body), n_cols, columns, preview) # --------------------------------------------------------------------------- # readers # --------------------------------------------------------------------------- def _resolve(columns: list[str], wanted: str | int | None, default_index: int | None) -> int | None: if wanted is None: return default_index if isinstance(wanted, int) or (isinstance(wanted, str) and wanted.isdigit()): return int(wanted) lowered = [c.lower() for c in columns] if wanted.lower() in lowered: return lowered.index(wanted.lower()) raise InputError([InputIssue("unknown_column", "error", f"Column {wanted!r} not found; available: {columns}")]) def _table(source, kind: str, file_label: str): text = _read_text(source) rows, delim, line_numbers = _split_table(text) if not rows: raise InputError([InputIssue("empty_file", "error", f"{file_label} is empty", file_label)]) sn = sniff(text, kind) header = rows[0] if sn.has_header else None body, lines = (rows[1:], line_numbers[1:]) if sn.has_header else (rows, line_numbers) return text, body, lines, sn, header def read_proteins_with_issues(source, mapping: dict | None = None, file_label: str = "proteins"): """-> (DataFrame[protein_id, sequence(, species)], issues). FASTA headers may carry species as ``OS=`` (UniProt style). Tables accept an optional species column (``mapping={'species': 'organism'}`` to choose it).""" mapping = mapping or {} issues: list[InputIssue] = [] text = _read_text(source) if is_fasta(text): ids, seqs, species, cur = [], [], [], [] for ln in text.split("\n"): if ln.startswith(">"): if ids: seqs.append("".join(cur)) head = ln[1:].strip() ids.append(head.split()[0] if head else "") m = re.search(r"\bOS=(.+?)(?:\s+[A-Z]{2}=|$)", head) species.append(m.group(1).strip() if m else "") cur = [] elif ln.strip(): cur.append(ln.strip()) if ids: seqs.append("".join(cur)) df = pd.DataFrame({"protein_id": ids, "sequence": seqs}) if any(species): df["species"] = species line_of = None else: _, body, lines, sn, _ = _table(text, "proteins", file_label) cols = sn.columns i_id = _resolve(cols, mapping.get("protein_id"), 0) i_seq = _resolve(cols, mapping.get("sequence"), 1) lowered = [c.lower() for c in cols] default_sp = next((lowered.index(n) for n in ("species", "organism", "taxon") if n in lowered), None) i_sp = _resolve(cols, mapping.get("species"), default_sp) short = [ln for r, ln in zip(body, lines) if len(r) <= max(i_id, i_seq)] if short: issues.append(InputIssue("missing_columns", "error", f"{len(short)} protein rows have fewer than 2 columns", file_label, short[:20], hint="Each row needs a protein identifier and its amino-acid sequence.")) raise InputError(issues) df = pd.DataFrame({"protein_id": [r[i_id] for r in body], "sequence": [r[i_seq] for r in body]}) if i_sp is not None: df["species"] = [r[i_sp] if i_sp < len(r) else "" for r in body] line_of = lines df["protein_id"] = df["protein_id"].astype(str).str.strip() df["sequence"] = df["sequence"].astype(str).str.strip().str.upper() def rows_of(mask): idx = np.where(mask)[0][:20] return [line_of[i] for i in idx] if line_of else [int(i) + 1 for i in idx] empty_id = (df["protein_id"] == "").to_numpy() if empty_id.any(): issues.append(InputIssue("empty_protein_id", "error", f"{int(empty_id.sum())} proteins have an empty identifier", file_label, rows_of(empty_id))) dup = df["protein_id"].duplicated(keep=False).to_numpy() & ~empty_id if dup.any(): issues.append(InputIssue("duplicate_protein_id", "error", f"{df.loc[dup, 'protein_id'].nunique()} protein identifiers occur more than once", file_label, rows_of(dup), sorted(df.loc[dup, "protein_id"].unique())[:10], "Each protein identifier must be unique; merge or rename duplicates.")) empty_seq = (df["sequence"].str.len() == 0).to_numpy() if empty_seq.any(): issues.append(InputIssue("missing_sequence", "error", f"{int(empty_seq.sum())} proteins have no sequence", file_label, rows_of(empty_seq), df.loc[empty_seq, "protein_id"].tolist()[:10])) non_letters = df["sequence"].apply(lambda s: bool(set(s) - SEQUENCE_CHARS)).to_numpy() if non_letters.any(): issues.append(InputIssue("invalid_sequence_characters", "error", f"{int(non_letters.sum())} sequences contain characters that are not amino-acid letters", file_label, rows_of(non_letters), df.loc[non_letters, "protein_id"].tolist()[:10], "Check the column mapping: the sequence column may be wrong.")) nonstd = df["sequence"].apply(lambda s: any(c not in AMINO_ACIDS for c in s)).to_numpy() & ~non_letters if nonstd.any(): issues.append(InputIssue("nonstandard_residues", "warning", f"{int(nonstd.sum())} sequences contain non-standard residues (e.g. X, U, B, Z, *); " "they are removed before descriptor computation", file_label, rows_of(nonstd), df.loc[nonstd, "protein_id"].tolist()[:10])) if "species" in df: df["species"] = df["species"].astype(str).str.strip() if (df["species"] == "").all(): df = df.drop(columns="species") return df.reset_index(drop=True), issues def read_pairs_with_issues(source, mapping: dict | None = None, file_label: str = "pairs"): """-> (DataFrame[protein_a, protein_b, label], issues). Without a label column every row is a positive and the framework constructs all negatives itself.""" mapping = mapping or {} issues: list[InputIssue] = [] _, body, lines, sn, _ = _table(source, "pairs", file_label) cols = sn.columns i_a = _resolve(cols, mapping.get("protein_a"), 0) i_b = _resolve(cols, mapping.get("protein_b"), 1) i_l = _resolve(cols, mapping.get("label"), 2 if sn.n_columns >= 3 else None) short = [ln for r, ln in zip(body, lines) if len(r) <= max(i_a, i_b)] if short: issues.append(InputIssue("missing_columns", "error", f"{len(short)} pair rows have fewer than 2 columns", file_label, short[:20], hint="Each row needs two protein identifiers.")) raise InputError(issues) df = pd.DataFrame({"protein_a": [r[i_a].strip() for r in body], "protein_b": [r[i_b].strip() for r in body]}) if i_l is None: df["label"] = 1 issues.append(InputIssue("no_label_column", "info", "No label column: every pair is treated as an observed interaction, and all negatives " "will be constructed by the framework (N1 random pairs become the reference set).", file_label)) else: raw = [r[i_l] if i_l < len(r) else "" for r in body] labels = [coerce_label(v) for v in raw] bad = [i for i, v in enumerate(labels) if v is None] if bad: issues.append(InputIssue("invalid_label", "error", f"{len(bad)} rows have a label that is not 0/1", file_label, [lines[i] for i in bad[:20]], sorted({raw[i] for i in bad})[:10], "Use 1 for an observed interaction and 0 for a supplied negative " "(true/false, yes/no and positive/negative are also accepted).")) raise InputError(issues) df["label"] = np.array(labels, dtype=int) if not (df["label"] == 0).any(): issues.append(InputIssue("no_negatives", "info", "No label-0 rows: all negatives will be constructed by the framework.", file_label)) empty = (df["protein_a"] == "") | (df["protein_b"] == "") if empty.any(): issues.append(InputIssue("empty_identifier", "error", f"{int(empty.sum())} pairs have an empty identifier", file_label, [lines[i] for i in np.where(empty)[0][:20]])) df.attrs["line_numbers"] = lines return df.reset_index(drop=True), issues def read_species_map(source, mapping: dict | None = None, file_label: str = "species") -> dict[str, str]: """``protein_idspecies`` table -> dict (blank species are dropped).""" mapping = mapping or {} _, body, _, sn, _ = _table(source, "species", file_label) cols = sn.columns i_id = _resolve(cols, mapping.get("protein_id"), 0) lowered = [c.lower() for c in cols] default_sp = next((lowered.index(n) for n in ("species", "organism", "taxon") if n in lowered), 1) i_sp = _resolve(cols, mapping.get("species"), default_sp) out = {} for r in body: if len(r) > max(i_id, i_sp) and r[i_sp].strip(): out[r[i_id].strip()] = r[i_sp].strip() return out def cross_reference_issues(proteins: pd.DataFrame, pairs: pd.DataFrame) -> list[InputIssue]: known = set(proteins["protein_id"]) lines = pairs.attrs.get("line_numbers") miss_a, miss_b = ~pairs["protein_a"].isin(known), ~pairs["protein_b"].isin(known) bad = (miss_a | miss_b).to_numpy() if not bad.any(): return [] missing = sorted(set(pairs.loc[miss_a, "protein_a"]) | set(pairs.loc[miss_b, "protein_b"])) rows = [lines[i] for i in np.where(bad)[0][:20]] if lines else [] return [InputIssue("unknown_identifier", "error", f"{len(missing)} identifiers in the pair file have no sequence ({int(bad.sum())} pairs affected)", "pairs", rows, missing[:10], "Every identifier in the pair file must appear in the protein file. Check for version " "suffixes (e.g. P12345.2) or different identifier namespaces.")] # Verbatim from ppi_benchmark_controls.py v1.0.0 ------------------------------- def _looks_like_header(first_row: list[str], expected: set[str]) -> bool: return any(str(x).strip().lower() in expected for x in first_row) def read_clusters(path) -> dict[str, str]: """Optional ``protein_idcluster_id`` table (e.g. MMseqs2 easy-cluster ``*_cluster.tsv`` output, which lists representativemember).""" raw = pd.read_csv(path, sep="\t", header=None, dtype=str) if _looks_like_header(raw.iloc[0].tolist(), {"protein_id", "cluster_id", "representative_id"}): header = [h.strip().lower() for h in raw.iloc[0]] raw = raw.iloc[1:] raw.columns = header return dict(zip(raw["protein_id"], raw["cluster_id"])) # headerless MMseqs2 layout: representative, member return dict(zip(raw.iloc[:, 1], raw.iloc[:, 0])) def read_embeddings(path, dataset: str = "mean") -> pd.DataFrame: """Optional per-protein embedding -> DataFrame[protein_id, f0..fN]. Formats: CSV/TSV (first column protein_id, remaining numeric), NPZ (arrays ``ids`` and ``X``), or HDF5 (datasets ``protein_id`` and ``mean``/``cls``).""" path = Path(path) suffix = path.suffix.lower() if suffix in {".h5", ".hdf5"}: import h5py with h5py.File(path, "r") as f: ids = [x.decode() if isinstance(x, bytes) else str(x) for x in f["protein_id"][:]] x = f[dataset][:] elif suffix == ".npz": data = np.load(path, allow_pickle=False) ids, x = [str(i) for i in data["ids"]], data["X"] else: sep = "," if suffix == ".csv" else "\t" df = pd.read_csv(path, sep=sep) ids, x = df.iloc[:, 0].astype(str).tolist(), df.iloc[:, 1:].to_numpy(dtype=np.float64) out = pd.DataFrame(np.asarray(x, dtype=np.float64), columns=[f"emb_{i}" for i in range(np.asarray(x).shape[1])]) out.insert(0, "protein_id", ids) return out def write_tsv_bytes(df: pd.DataFrame) -> bytes: buf = _io.StringIO() df.to_csv(buf, sep="\t", index=False) return buf.getvalue().encode()