avoigt1121
Claude Opus 5.5
feat: the five knowledge endpoints, gradio_ui and frozen descriptor (s2)
3ed6b66 Download src/identifiers.py from anne-voigt/pdac-knowledge-agent-dev: direct link, hf CLI and curl.
- Browser
- Download file 6.4 kB
-
https://huggingface.co/spaces/anne-voigt/pdac-knowledge-agent-dev/resolve/main/src/identifiers.py
- Command line
-
hf download hf://spaces/anne-voigt/pdac-knowledge-agent-dev/src/identifiers.py
-
curl -L -o identifiers.py https://huggingface.co/spaces/anne-voigt/pdac-knowledge-agent-dev/resolve/main/src/identifiers.py
6.4 kB
| """What may leave the Space: identifiers only (ADR-0001 §6). | |
| Every endpoint argument passes through here BEFORE any outbound call. Allowed are HGNC symbols, | |
| Ensembl gene ids, UniProt accession numbers, disease ids from EFO, MONDO or Orphanet, NCT ids, and | |
| a disease name as a bounded string. Free text, sample ids, numbers (expression values) and file contents are | |
| refused as data — ``invalid_identifier`` — and more than 100 identifiers as | |
| ``too_many_identifiers``. A refusal never echoes more than a short, truncated token back. | |
| """ | |
| from __future__ import annotations | |
| import re | |
| from dataclasses import dataclass, field | |
| MAX_IDENTIFIERS = 100 | |
| MAX_DISEASE_NAME = 80 | |
| _ECHO = 24 # longest token a refusal message repeats back | |
| # HGNC approved symbols start with a letter and use letters, digits and hyphens (C1orf112, | |
| # HLA-DRB1, NKX2-1; immunoglobulin segments such as IGHV1-69-2 carry two hyphens). | |
| SYMBOL = re.compile(r"^[A-Za-z][A-Za-z0-9]*(?:-[A-Za-z0-9]+){0,2}$") | |
| ENSEMBL_GENE = re.compile(r"^ENSG\d{11}(?:\.\d+)?$") | |
| # UniProt's published accession pattern. | |
| UNIPROT = re.compile( | |
| r"^(?:[OPQ][0-9][A-Z0-9]{3}[0-9]|[A-NR-Z][0-9](?:[A-Z][A-Z0-9]{2}[0-9]){1,2})$" | |
| ) | |
| DISEASE_ID = re.compile(r"^(?:EFO|MONDO|Orphanet)_\d{1,10}$") | |
| NCT = re.compile(r"^NCT\d{8}$") | |
| DISEASE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 ,'\-]*$") | |
| # Shapes that would pass as a symbol but are sample or run identifiers (TCGA barcodes, GEO and | |
| # SRA accessions, CPTAC case ids, ICGC donor/sample ids). They never leave the Space: a sample id | |
| # is study data, not a public identifier. Every prefix here is checked against real symbols that | |
| # share its letters (CPT1A, ERRFI1, DOK1, SAA1 stay valid; tests/test_identifiers.py). | |
| _SAMPLE_ID = re.compile( | |
| r"^(?:TCGA-|GSM\d|GSE\d|SRR\d|ERR\d|DRR\d|SRS\d|SAMN\d|C3[LN]-|DO\d|SA\d)", | |
| re.IGNORECASE, | |
| ) | |
| _SPLIT = re.compile(r"[\s,;]+") | |
| GENE_KINDS = frozenset({"symbol", "ensembl"}) | |
| PROTEIN_OR_GENE_KINDS = frozenset({"symbol", "ensembl", "uniprot"}) | |
| class Identifier: | |
| value: str | |
| kind: str # symbol | ensembl | uniprot | disease_id | disease_name | nct | |
| class Parsed: | |
| """Either a list of identifiers or a refusal (``reason`` + ``message``), never both.""" | |
| identifiers: tuple[Identifier, ...] = () | |
| reason: str | None = None | |
| message: str | None = None | |
| duplicates_dropped: int = field(default=0) | |
| def ok(self) -> bool: | |
| return self.reason is None | |
| def _echo(token: str) -> str: | |
| return token if len(token) <= _ECHO else token[:_ECHO] + "…" | |
| def _refuse(reason: str, message: str) -> Parsed: | |
| return Parsed(reason=reason, message=message) | |
| def _dedupe_key(token: str) -> str: | |
| """Symbols compare case-insensitively (kras == KRAS); Ensembl ids exactly.""" | |
| return token if token.startswith("ENSG") else token.upper() | |
| def classify_gene_token( | |
| token: str, *, allow_uniprot: bool = False, allow_ensembl: bool = True | |
| ) -> str | None: | |
| """The identifier kind for one gene/protein token, or None if it is not one.""" | |
| if _SAMPLE_ID.match(token): | |
| return None | |
| if ENSEMBL_GENE.match(token): | |
| return "ensembl" if allow_ensembl else None | |
| if allow_uniprot and UNIPROT.match(token): | |
| return "uniprot" | |
| if len(token) <= 25 and SYMBOL.match(token): | |
| return "symbol" | |
| return None | |
| def parse_gene_list( | |
| raw: str | list[str] | None, *, allow_uniprot: bool = False, allow_ensembl: bool = True | |
| ) -> Parsed: | |
| """Split, de-duplicate (order kept) and validate a gene list. | |
| ``allow_ensembl=False`` is for the endpoints that resolve through HGNC symbols | |
| (annotate_genes, pathways_for_genes); drugs_for_targets accepts Ensembl ids directly. | |
| """ | |
| if isinstance(raw, list): | |
| tokens = [str(t).strip() for t in raw] | |
| else: | |
| tokens = _SPLIT.split((raw or "").strip()) | |
| tokens = [t for t in tokens if t] | |
| if not tokens: | |
| return _refuse("invalid_identifier", "No gene identifiers were given.") | |
| unique = {_dedupe_key(t) for t in tokens} | |
| if len(unique) > MAX_IDENTIFIERS: | |
| return _refuse( | |
| "too_many_identifiers", | |
| f"{len(unique)} identifiers given; at most {MAX_IDENTIFIERS} per call.", | |
| ) | |
| kinds = {"allow_uniprot": allow_uniprot, "allow_ensembl": allow_ensembl} | |
| bad = [t for t in tokens if classify_gene_token(t, **kinds) is None] | |
| if bad: | |
| shown = ", ".join(_echo(t) for t in bad[:5]) | |
| more = f" and {len(bad) - 5} more" if len(bad) > 5 else "" | |
| accepted = ( | |
| "HGNC symbols" | |
| + (" or Ensembl gene ids" if allow_ensembl else "") | |
| + (" or UniProt accessions" if allow_uniprot else "") | |
| ) | |
| return _refuse( | |
| "invalid_identifier", | |
| f"Not a gene identifier: {shown}{more}. Only {accepted} are sent to external sources; " | |
| "free text, sample ids and numbers are not.", | |
| ) | |
| out: list[Identifier] = [] | |
| emitted: set[str] = set() | |
| for t in tokens: | |
| key = _dedupe_key(t) | |
| if key in emitted: | |
| continue | |
| emitted.add(key) | |
| out.append(Identifier(t, classify_gene_token(t, **kinds) or "symbol")) | |
| return Parsed(identifiers=tuple(out), duplicates_dropped=len(tokens) - len(out)) | |
| def parse_disease(raw: str | None) -> Parsed: | |
| """One disease: an EFO/MONDO/Orphanet id, or a bounded name.""" | |
| text = (raw or "").strip() | |
| if not text: | |
| return _refuse("invalid_identifier", "No disease was given.") | |
| if DISEASE_ID.match(text): | |
| return Parsed(identifiers=(Identifier(text, "disease_id"),)) | |
| if ( | |
| len(text) <= MAX_DISEASE_NAME | |
| and DISEASE_NAME.match(text) | |
| and any(c.isalpha() for c in text) | |
| and not _SAMPLE_ID.match(text) | |
| ): | |
| return Parsed(identifiers=(Identifier(" ".join(text.split()), "disease_name"),)) | |
| return _refuse( | |
| "invalid_identifier", | |
| f"Not a disease name or id: {_echo(text)}. Give an EFO/MONDO id or a name of at most " | |
| f"{MAX_DISEASE_NAME} characters (letters, digits, spaces, hyphens, commas, apostrophes).", | |
| ) | |
| def parse_nct(raw: str | None) -> Parsed: | |
| text = (raw or "").strip().upper() | |
| if NCT.match(text): | |
| return Parsed(identifiers=(Identifier(text, "nct"),)) | |
| return _refuse("invalid_identifier", f"Not an NCT id: {_echo(text)}.") | |