From 5c330c155352e1db5bfa491dcdf919ad7f41fc2a Mon Sep 17 00:00:00 2001 From: wanieldd <309034407+wanieldd@users.noreply.github.com> Date: Wed, 26 Aug 2026 13:46:03 -0600 Subject: [PATCH] tests: address issue #16 (fixes #16) --- README.md | 106 ++++++++++- scibase/__init__.py | 44 +++++ scibase/gap_finder.py | 267 ++++++++++++++++++++++++++ scibase/reproducibility.py | 257 +++++++++++++++++++++++++ scibase/review.py | 350 ++++++++++++++++++++++++++++++++++ tests/test_scibase.py | 374 +++++++++++++++++++++++++++++++++++++ 6 files changed, 1396 insertions(+), 2 deletions(-) create mode 100644 scibase/__init__.py create mode 100644 scibase/gap_finder.py create mode 100644 scibase/reproducibility.py create mode 100644 scibase/review.py create mode 100644 tests/test_scibase.py diff --git a/README.md b/README.md index d338cf68..39d9d564 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,104 @@ -# deepevents.ai -deepevents.ai main codebase +# AI-Powered Research Assistant Suite + +A self-contained toolkit that augments scientific workflows with an embedded +analyst, reviewer, and strategist. The suite is dependency-free (Python +standard library only) and ships with a full pytest suite. + +## Capabilities + +### 1. Auto Peer Review Reports — `scibase.review` + +Analyzes a manuscript and generates structured, category-based review +suggestions: + +- **Clarity and coherence** checks (jargon, vague wording, sentence length) +- **Statistical red flags** (unreported p-values, sample sizes, correlations) +- **Methodological red flags** (e.g. missing survey response rates) +- **Missing citations** and scope misalignment +- **Claims vs. evidence alignment** (unsupported or thinly-evidenced claims) + +Templates are adaptable per domain (`molecular-biology`, `quantum-physics`, +`clinical-trials`); unknown domains fall back to a generic template. Each +review produces a 0-100 score. + +```python +from scibase import Manuscript, generate_peer_review + +ms = Manuscript( + title="My study", + abstract="A significant p-value was observed.", + claims=[("The treatment works.", None)], + citations=["Smith et al., 2021"], + domain="clinical-trials", +) +report = generate_peer_review(ms) +print(report.summary) +print(report.score) # 0-100 +for issue in report.issues: + print(issue.category, issue.severity, issue.message, issue.suggestion) +``` + +### 2. Reproducibility Checker — `scibase.reproducibility` + +Inspects a project directory and verifies: + +- **Pipeline presence** — source/notebook files exist +- **Raw data present** — data files under `data/` +- **Tests present** — `test_*` files +- **Output consistency** — reported results match actual file contents +- **Dependency/version integrity** — every dependency is pinned +- **Determinism** — byte-identical outputs across `reproducibility/runs/` + +Each check contributes a reproducibility confidence score from 0.0 to 1.0. + +```python +from pathlib import Path +from scibase import Project, ReproducibilityChecker + +project = Project( + name="example", + root=Path("my_research"), + reported_results={"results.txt": "accuracy=0.94"}, + dependencies={"numpy": "1.26.4", "pandas": None}, # None => unpinned +) +report = ReproducibilityChecker().check(project) +print(report.score, report.reproducible) +for issue in report.issues: + print(issue.code, issue.message) +``` + +### 3. Research Gap Finder — `scibase.gap_finder` + +Scans a corpus of papers and identifies: + +- **Under-studied intersections** — topic combinations where each topic is + individually active but the combination rarely (or never) co-occurs +- **Frequently cited unresolved questions** — extracted from limitations + sections, ranked by citation count +- A personalized **research opportunities feed** ranked against the user's + interests and project history + +```python +from scibase import GapFinder, Paper + +papers = [ + Paper(id="p1", title="...", topics=["CRISPR", "Alzheimer's"], citations=40, + open_questions=["The role of glia remains unclear."]), + # ... +] +finder = GapFinder(papers) +for opp in finder.under_studied_intersections(): + print(opp.topics, opp.rationale, opp.score) + +for question, papers, citations in finder.unresolved_questions(): + print(question, papers, citations) + +feed = finder.research_opportunities_feed( + interests=["single-cell RNA-seq"], project_history=["CRISPR"]) +``` + +## Running the tests + +```bash +python -m pytest tests/ -q +``` \ No newline at end of file diff --git a/scibase/__init__.py b/scibase/__init__.py new file mode 100644 index 00000000..9cd35095 --- /dev/null +++ b/scibase/__init__.py @@ -0,0 +1,44 @@ +"""AI-Powered Research Assistant Suite. + +A self-contained toolkit that augments scientific workflows with: + +- Auto peer review reports (:mod:`scibase.review`) +- A reproducibility checker (:mod:`scibase.reproducibility`) +- A research gap finder (:mod:`scibase.gap_finder`) + +All modules use only the Python standard library so the suite runs anywhere. +""" + +from .gap_finder import GapFinder, Paper, ResearchOpportunity +from .reproducibility import ( + Project, + ReproducibilityIssue, + ReproducibilityReport, + ReproducibilityChecker, +) +from .review import ( + Manuscript, + ReviewCategory, + ReviewIssue, + ReviewReport, + generate_peer_review, + peer_review_score, +) + +__all__ = [ + "GapFinder", + "Manuscript", + "Paper", + "Project", + "ReproducibilityChecker", + "ReproducibilityIssue", + "ReproducibilityReport", + "ResearchOpportunity", + "ReviewCategory", + "ReviewIssue", + "ReviewReport", + "generate_peer_review", + "peer_review_score", +] + +__version__ = "0.1.0" \ No newline at end of file diff --git a/scibase/gap_finder.py b/scibase/gap_finder.py new file mode 100644 index 00000000..178ca554 --- /dev/null +++ b/scibase/gap_finder.py @@ -0,0 +1,267 @@ +"""Research gap finder. + +Scans a corpus of papers and identifies under-studied topic intersections, +frequently cited unresolved questions, and generates a "research +opportunities" feed personalized for a user's interests and project history. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field + +MIN_POPULARITY = 3 # a topic must appear this many times to count as "active" +MIN_INTEREST_OVERLAP = 1 # opportunities must share this many user interests + + +@dataclass +class Paper: + """A paper in the scanned corpus. + + Parameters + ---------- + id: + Stable paper identifier. + title: + Paper title. + abstract: + Paper abstract. + topics: + Topics covered by the paper. + citations: + Number of times the paper has been cited (published + in-progress). + open_questions: + Explicit unresolved questions stated in the paper (e.g. limitations + sections that suggest open directions). + """ + + id: str + title: str + abstract: str = "" + topics: list[str] = field(default_factory=list) + citations: int = 0 + open_questions: list[str] = field(default_factory=list) + + +@dataclass +class ResearchOpportunity: + """A suggested research direction derived from corpus gaps. + + Parameters + ---------- + topics: + The under-studied topic combination. + popularity: + How often each member topic appears in the corpus. + combination_count: + How many papers cover the full combination. + rationale: + Human-readable justification for the suggestion. + relevance: + Number of user interests / project topics the opportunity overlaps. + """ + + topics: frozenset[str] + popularity: dict[str, int] + combination_count: int + rationale: str + relevance: int = 0 + + @property + def score(self) -> float: + """Recommendation score: active topics, low replication, high relevance.""" + activity = sum(self.popularity.values()) + return round((activity * (1 + self.relevance)) / (1 + self.combination_count), 2) + + +class GapFinder: + """Identifies research gaps across a corpus of :class:`Paper` objects.""" + + def __init__(self, papers: list[Paper] | None = None) -> None: + self.papers: list[Paper] = papers or [] + + def add(self, paper: Paper) -> None: + """Add a paper to the corpus.""" + self.papers.append(paper) + + def add_papers(self, papers: list[Paper]) -> None: + """Add several papers to the corpus.""" + self.papers.extend(papers) + + # Public API --------------------------------------------------------------- + + def topic_popularity(self) -> dict[str, int]: + """Count how many papers cover each topic.""" + counts: dict[str, int] = {} + for paper in self.papers: + for topic in paper.topics: + counts[topic] = counts.get(topic, 0) + 1 + return counts + + def under_studied_intersections(self, size: int = 2) -> list[ResearchOpportunity]: + """Find topic combinations with high activity but low replication. + + A combination is considered a gap when every member topic is + individually active (appears at least :data:`MIN_POPULARITY` times) + but the combination itself is covered by very few papers. Pairs where + every paper already covers all members are excluded. + + Parameters + ---------- + size: + Number of topics per combination. + + Returns + ------- + list[ResearchOpportunity] + Opportunities sorted by recommendation score (descending). + """ + popularity = self.topic_popularity() + active = sorted(t for t, n in popularity.items() if n >= MIN_POPULARITY) + + from itertools import combinations + + opportunities: list[ResearchOpportunity] = [] + for combo in combinations(active, size): + combo_set = frozenset(combo) + covering = [ + paper + for paper in self.papers + if combo_set.issubset(set(paper.topics)) + ] + combo_popularity = {t: popularity[t] for t in combo} + if not covering: + rationale = ( + f"Topics {', '.join(combo)} are individually active but never co-occur in the corpus." + ) + else: + combo_count = len(covering) + fully_covered = sum(1 for p in covering if set(p.topics) <= combo_set) + if fully_covered == combo_count: + continue # every paper on these topics already covers the full intersection + rationale = ( + f"Topics {', '.join(combo)} co-occur in only {combo_count} paper(s) " + "despite high individual activity." + ) + opportunities.append( + ResearchOpportunity( + topics=combo_set, + popularity=combo_popularity, + combination_count=len(covering), + rationale=rationale, + ) + ) + + opportunities.sort(key=lambda o: o.score, reverse=True) + return opportunities + + def unresolved_questions(self, min_citations: int = 5) -> list[tuple[str, int, int]]: + """Return frequently cited unresolved questions. + + Questions are extracted from each paper's ``open_questions`` list and + aggregated. Only questions appearing in papers with at least + ``min_citations`` citations are returned. + + Parameters + ---------- + min_citations: + Minimum citation count for a paper's questions to be included. + + Returns + ------- + list[tuple[str, int, int]] + ``(question, paper_count, total_citations)`` sorted by total + citations (descending). + """ + questions: dict[str, list[int]] = {} + for paper in self.papers: + if paper.citations < min_citations: + continue + for question in paper.open_questions: + key = question.strip().rstrip("?") + if not key: + continue + questions.setdefault(key, []).append(paper.citations) + + result = [ + (question, len(cites), sum(cites)) + for question, cites in questions.items() + ] + result.sort(key=lambda item: item[2], reverse=True) + return result + + def research_opportunities_feed( + self, + interests: list[str], + project_history: list[str] | None = None, + size: int = 2, + ) -> list[ResearchOpportunity]: + """Generate a personalized "research opportunities" feed. + + Under-studied intersections are ranked by overlap with the user's + interests and project history. An opportunity must overlap at least + :data:`MIN_INTEREST_OVERLAP` interest/project topic to be included. + + Parameters + ---------- + interests: + Topics the user is interested in. + project_history: + Topics from the user's past projects. + size: + Number of topics per intersection. + + Returns + ------- + list[ResearchOpportunity] + Personalized opportunities sorted by recommendation score. + """ + context = set(interests) | set(project_history or []) + opportunities = self.under_studied_intersections(size=size) + + feed: list[ResearchOpportunity] = [] + for opportunity in opportunities: + relevance = len(opportunity.topics & context) + if relevance < MIN_INTEREST_OVERLAP: + continue + feed.append( + ResearchOpportunity( + topics=opportunity.topics, + popularity=opportunity.popularity, + combination_count=opportunity.combination_count, + rationale=opportunity.rationale, + relevance=relevance, + ) + ) + + feed.sort(key=lambda o: o.score, reverse=True) + return feed + + +def extract_questions(text: str) -> list[str]: + """Extract explicit unresolved-question sentences from a text. + + Looks for sentences containing interrogative phrasing typical of + limitations sections ("remains unclear", "future work", "open question", + etc.) or ending with a question mark. + + Parameters + ---------- + text: + Text to scan (e.g. a paper's limitations section). + + Returns + ------- + list[str] + Extracted question sentences. + """ + sentences = re.split(r"(?<=[.!?])\s+", text.strip()) + found: list[str] = [] + for sentence in sentences: + lowered = sentence.lower() + if "?" in sentence or any( + marker in lowered + for marker in ("remains unclear", "open question", "future work", "not well understood", "requires further") + ): + found.append(sentence.strip()) + return found \ No newline at end of file diff --git a/scibase/reproducibility.py b/scibase/reproducibility.py new file mode 100644 index 00000000..a5b874a5 --- /dev/null +++ b/scibase/reproducibility.py @@ -0,0 +1,257 @@ +"""Reproducibility checker. + +Inspects a project directory to verify output consistency with reported +results, dependency/version integrity, and the presence of raw data, clean +pipelines, and test sets. Flags discrepancies or non-determinism and assigns a +reproducibility confidence score. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path + +PIPELINE_EXTENSIONS = {".py", ".ipynb", ".r", ".jl"} +DATA_EXTENSIONS = {".csv", ".tsv", ".json", ".jsonl", ".parquet", ".h5", ".hdf5", ".npy", ".npz"} + + +@dataclass +class ReproducibilityIssue: + """A single discrepancy found by the checker. + + Parameters + ---------- + code: + Stable machine-readable identifier for the issue. + message: + Human-readable description of the problem. + """ + + code: str + message: str + + +@dataclass +class Project: + """A research project to check for reproducibility. + + Parameters + ---------- + name: + Project name. + root: + Directory containing the project files. + reported_results: + Optional mapping of output name -> reported value (as string). When + an output file is present, the checker compares its contents against + the reported value. + dependencies: + Optional mapping of dependency name -> version. Dependencies without + a version are flagged as unpinned. + """ + + name: str + root: Path + reported_results: dict[str, str] = field(default_factory=dict) + dependencies: dict[str, str | None] = field(default_factory=dict) + + +@dataclass +class ReproducibilityReport: + """Result of running the reproducibility checker over a :class:`Project`. + + Attributes + ---------- + project: + The project that was checked. + issues: + Discrepancies and missing requirements found. + checks_passed: + Names of the checks that passed. + score: + Reproducibility confidence score from 0.0 to 1.0. + """ + + project: Project + issues: list[ReproducibilityIssue] = field(default_factory=list) + checks_passed: list[str] = field(default_factory=list) + score: float = 1.0 + + @property + def reproducible(self) -> bool: + """True when every hard requirement passes (score above 0.5).""" + return self.score >= 0.5 + + +class ReproducibilityChecker: + """Runs reproducibility checks against a :class:`Project` directory.""" + + def check(self, project: Project) -> ReproducibilityReport: + """Check ``project`` and return a :class:`ReproducibilityReport`. + + The following checks are performed: + + - **Pipeline presence**: at least one source or notebook file exists. + - **Raw data present**: at least one raw data file exists under a + ``data/`` directory. + - **Tests present**: at least one test file exists. + - **Output consistency**: when a reported result names a file, the + file must exist and its contents must match the reported value. + - **Dependency integrity**: every dependency must have a pinned + version. + - **Determinism**: when a ``reproducibility/runs/`` directory exists, + each output file must be byte-identical across runs. + + Parameters + ---------- + project: + The project to check. + + Returns + ------- + ReproducibilityReport + Issues found plus a confidence score in ``[0.0, 1.0]``. + """ + root = project.root + issues: list[ReproducibilityIssue] = [] + passed: list[str] = [] + + pipeline_files = self._find_pipeline_files(root) + if pipeline_files: + passed.append("pipeline") + else: + issues.append( + ReproducibilityIssue( + "no-pipeline", + "No source or notebook files found; a clean pipeline is required.", + ) + ) + + data_files = self._find_data_files(root) + if data_files: + passed.append("raw-data") + else: + issues.append( + ReproducibilityIssue( + "no-raw-data", + "No raw data files found under a data/ directory.", + ) + ) + + test_files = self._find_test_files(root) + if test_files: + passed.append("tests") + else: + issues.append( + ReproducibilityIssue( + "no-tests", + "No test files found.", + ) + ) + + issues.extend(self._check_output_consistency(project)) + if not any(i.code == "output-mismatch" for i in issues): + passed.append("output-consistency") + + issues.extend(self._check_dependencies(project)) + if not any(i.code == "unpinned-dependency" for i in issues): + passed.append("dependencies") + + issues.extend(self._check_determinism(root)) + if not any(i.code == "non-deterministic" for i in issues): + passed.append("determinism") + + score = self._score(issues) + return ReproducibilityReport(project=project, issues=issues, checks_passed=passed, score=score) + + # Internal helpers --------------------------------------------------------- + + @staticmethod + def _find_pipeline_files(root: Path) -> list[Path]: + if not root.exists(): + return [] + return [p for p in root.rglob("*") if p.is_file() and p.suffix.lower() in PIPELINE_EXTENSIONS] + + @staticmethod + def _find_data_files(root: Path) -> list[Path]: + if not root.exists(): + return [] + data_root = root / "data" + if not data_root.exists(): + return [] + return [p for p in data_root.rglob("*") if p.is_file() and p.suffix.lower() in DATA_EXTENSIONS] + + @staticmethod + def _find_test_files(root: Path) -> list[Path]: + if not root.exists(): + return [] + return [p for p in root.rglob("*") if p.is_file() and (p.name.startswith("test_") or p.name.startswith("tests/"))] + + def _check_output_consistency(self, project: Project) -> list[ReproducibilityIssue]: + issues: list[ReproducibilityIssue] = [] + for output_name, reported in project.reported_results.items(): + output_file = project.root / output_name + if not output_file.exists(): + issues.append( + ReproducibilityIssue( + "missing-output", + f"Reported output {output_name!r} does not exist.", + ) + ) + continue + actual = output_file.read_text(encoding="utf-8").strip() + if actual != reported.strip(): + issues.append( + ReproducibilityIssue( + "output-mismatch", + ( + f"Reported output {output_name!r} does not match the file contents " + f"(reported {reported.strip()!r}, found {actual!r})." + ), + ) + ) + return issues + + def _check_dependencies(self, project: Project) -> list[ReproducibilityIssue]: + issues: list[ReproducibilityIssue] = [] + for dep, version in project.dependencies.items(): + if not version: + issues.append( + ReproducibilityIssue( + "unpinned-dependency", + f"Dependency {dep!r} is not pinned to a version.", + ) + ) + return issues + + def _check_determinism(self, root: Path) -> list[ReproducibilityIssue]: + issues: list[ReproducibilityIssue] = [] + runs_dir = root / "reproducibility" / "runs" + if not runs_dir.exists(): + return issues + + run_outputs: dict[str, set[bytes]] = {} + for run in sorted(runs_dir.iterdir()): + if not run.is_dir(): + continue + for output in run.rglob("*"): + if not output.is_file(): + continue + run_outputs.setdefault(output.name, set()).add(output.read_bytes()) + + for output_name, contents in run_outputs.items(): + if len(contents) > 1: + issues.append( + ReproducibilityIssue( + "non-deterministic", + f"Output {output_name!r} differs across reproducibility runs.", + ) + ) + return issues + + @staticmethod + def _score(issues: list[ReproducibilityIssue]) -> float: + if not issues: + return 1.0 + deduction = 0.2 * len(issues) + return round(max(0.0, min(1.0, 1.0 - deduction)), 2) \ No newline at end of file diff --git a/scibase/review.py b/scibase/review.py new file mode 100644 index 00000000..e5824f1b --- /dev/null +++ b/scibase/review.py @@ -0,0 +1,350 @@ +"""Auto peer review report generation. + +Analyzes a manuscript and produces structured, category-based review +suggestions covering clarity and coherence, statistical and methodological +red flags, missing citations, and claims-vs-evidence alignment. Templates are +adaptable per research domain. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from enum import Enum + +SEVERITY_WEIGHTS = {"error": 10, "warning": 4, "info": 1} + +DEFAULT_DOMAIN_TEMPLATES: dict[str, dict[str, str]] = { + "molecular-biology": { + "name": "Molecular Biology", + "intro": "Review guided by molecular biology reporting standards.", + "checks": "Verify cell lines, replicate counts, and reagent lot numbers.", + }, + "quantum-physics": { + "name": "Quantum Physics", + "intro": "Review guided by quantum physics reporting standards.", + "checks": "Verify gate fidelities, decoherence times, and error bars.", + }, + "clinical-trials": { + "name": "Clinical Trials", + "intro": "Review guided by clinical trial reporting standards.", + "checks": "Verify randomization, blinding, and adverse-event reporting.", + }, +} + + +class ReviewCategory(str, Enum): + """Categories of review feedback produced by :func:`generate_peer_review`.""" + + CLARITY = "clarity" + STATISTICAL = "statistical" + METHODOLOGY = "methodology" + CITATION = "citations" + CLAIMS = "claims" + SCOPE = "scope" + + +@dataclass +class Manuscript: + """The document being reviewed. + + Parameters + ---------- + title: + Manuscript title. + abstract: + Abstract text (subject to clarity/claims checks). + body: + Main manuscript text (subject to statistical/methodological checks). + claims: + Claims stated by the authors as ``(assertion, evidence)`` pairs. + ``evidence`` may be ``None`` when the claim is unsupported. + citations: + References cited in the manuscript. + domain: + Optional domain key (e.g. ``"clinical-trials"``) used to select an + adaptive review template. Unknown domains fall back to a generic + template. + """ + + title: str + abstract: str + body: str = "" + claims: list[tuple[str, str | None]] = field(default_factory=list) + citations: list[str] = field(default_factory=list) + domain: str | None = None + + +@dataclass +class ReviewIssue: + """A single structured review suggestion.""" + + category: ReviewCategory + severity: str + location: str + message: str + suggestion: str + + @property + def weight(self) -> int: + """Severity weight used when scoring the report.""" + return SEVERITY_WEIGHTS.get(self.severity, 0) + + +@dataclass +class ReviewReport: + """Structured output of a peer review pass over a manuscript.""" + + title: str + domain: str | None + issues: list[ReviewIssue] = field(default_factory=list) + summary: str = "" + + @property + def score(self) -> int: + """Peer review score on a 0-100 scale (higher is better).""" + if not self.issues: + return 100 + deduction = sum(issue.weight for issue in self.issues) + return max(0, min(100, 100 - deduction)) + + +# Heuristic rules ----------------------------------------------------------------- + + +def _find_clarity_issues(manuscript: Manuscript) -> list[ReviewIssue]: + issues: list[ReviewIssue] = [] + text = f"{manuscript.abstract}\n{manuscript.body}" + + if len(text.split()) < 40: + issues.append( + ReviewIssue( + category=ReviewCategory.CLARITY, + severity="warning", + location="manuscript", + message="The manuscript body is very short and may lack sufficient detail.", + suggestion="Expand the manuscript with the methods, results, and discussion sections.", + ) + ) + + jargon = re.findall(r"\b(?:etc\.|very|extremely|significantly|novel|robust)\b", text, re.IGNORECASE) + if jargon: + issues.append( + ReviewIssue( + category=ReviewCategory.CLARITY, + severity="info", + location="manuscript", + message=( + "Vague or overused wording detected " + f"(e.g. {', '.join(dict.fromkeys(j.lower() for j in jargon))})." + ), + suggestion="Replace vague qualifiers with concrete measurements or precise language.", + ) + ) + + sentences = re.split(r"[.!?]\s+", text) + long = [s for s in sentences if len(s.split()) > 60] + if long: + issues.append( + ReviewIssue( + category=ReviewCategory.CLARITY, + severity="warning", + location="manuscript", + message=f"{len(long)} sentence(s) exceed 60 words and harm readability.", + suggestion="Split long sentences to improve readability and coherence.", + ) + ) + + return issues + + +def _find_statistical_issues(manuscript: Manuscript) -> list[ReviewIssue]: + issues: list[ReviewIssue] = [] + text = f"{manuscript.abstract}\n{manuscript.body}".lower() + + if "p-value" in text or "p value" in text or "p<" in text or "p <" in text: + if not re.search(r"p\s*[<>=]\s*0?\.?\d", text): + issues.append( + ReviewIssue( + category=ReviewCategory.STATISTICAL, + severity="error", + location="abstract/body", + message="A p-value is mentioned without a reported numerical value.", + suggestion="Report the exact p-value (e.g. p = 0.023) alongside the test used.", + ) + ) + + if re.search(r"\bsample size\b", text): + if not re.search(r"\bsample size\b[^\n]{0,120}\d", text): + issues.append( + ReviewIssue( + category=ReviewCategory.STATISTICAL, + severity="warning", + location="methods", + message="Sample size is mentioned without a reported number.", + suggestion="State the exact sample size and the power analysis that justified it.", + ) + ) + + if re.search(r"\bcorrelat", text) and not re.search(r"\br\s*[=]\s*-?\d", text): + issues.append( + ReviewIssue( + category=ReviewCategory.STATISTICAL, + severity="warning", + location="results", + message="A correlation is claimed without reporting the coefficient.", + suggestion="Report the correlation coefficient (r) and its confidence interval.", + ) + ) + + return issues + + +def _find_methodological_issues(manuscript: Manuscript) -> list[ReviewIssue]: + issues: list[ReviewIssue] = [] + text = f"{manuscript.abstract}\n{manuscript.body}".lower() + + if re.search(r"\bsurvey\b", text) and not re.search(r"\bresponse rate\b", text): + issues.append( + ReviewIssue( + category=ReviewCategory.METHODOLOGY, + severity="warning", + location="methods", + message="A survey-based study is reported without a response rate.", + suggestion="Report the response rate and sampling strategy.", + ) + ) + + return issues + + +def _find_citation_issues(manuscript: Manuscript) -> list[ReviewIssue]: + issues: list[ReviewIssue] = [] + text = f"{manuscript.abstract}\n{manuscript.body}" + + if not manuscript.citations: + issues.append( + ReviewIssue( + category=ReviewCategory.CITATION, + severity="error", + location="manuscript", + message="No citations were provided for the manuscript.", + suggestion="Add citations supporting key claims and prior work.", + ) + ) + + # A claim is expected to reference prior literature; if the body uses + # "shown to" / "previous work" but cites nothing, flag scope alignment. + if manuscript.citations and re.search(r"\bshown to\b|\bprevious work\b|\bas reported\b", text) and len(manuscript.citations) < 2: + issues.append( + ReviewIssue( + category=ReviewCategory.CITATION, + severity="warning", + location="introduction", + message="The manuscript references prior work but cites very few sources.", + suggestion="Add citations for each reference to previous work.", + ) + ) + + return issues + + +def _find_claims_issues(manuscript: Manuscript) -> list[ReviewIssue]: + issues: list[ReviewIssue] = [] + + for idx, (claim, evidence) in enumerate(manuscript.claims, start=1): + location = f"claim #{idx}" + if evidence is None or not evidence.strip(): + issues.append( + ReviewIssue( + category=ReviewCategory.CLAIMS, + severity="error", + location=location, + message=f"Claim is not supported by evidence: {claim!r}", + suggestion="Provide supporting data, analysis, or a citation for this claim.", + ) + ) + elif len(evidence.split()) < 3: + issues.append( + ReviewIssue( + category=ReviewCategory.CLAIMS, + severity="warning", + location=location, + message=f"Evidence for claim is too thin: {claim!r}", + suggestion="Strengthen the evidence with quantitative results or citations.", + ) + ) + + return issues + + +def _resolve_domain(domain: str | None) -> dict[str, str] | None: + if domain is None: + return None + return DEFAULT_DOMAIN_TEMPLATES.get(domain.lower()) + + +# Public API ---------------------------------------------------------------------- + + +def generate_peer_review(manuscript: Manuscript) -> ReviewReport: + """Generate a structured peer review report for ``manuscript``. + + The report contains categorized :class:`ReviewIssue` objects covering + clarity and coherence, statistical and methodological red flags, missing + citations, and claims-vs-evidence alignment. When ``manuscript.domain`` + matches a known domain, an adaptive template is applied and surfaced in + the report summary. + + Parameters + ---------- + manuscript: + The manuscript under review. + + Returns + ------- + ReviewReport + Structured review suggestions plus an overall score. + """ + issues: list[ReviewIssue] = [] + issues.extend(_find_clarity_issues(manuscript)) + issues.extend(_find_statistical_issues(manuscript)) + issues.extend(_find_methodological_issues(manuscript)) + issues.extend(_find_citation_issues(manuscript)) + issues.extend(_find_claims_issues(manuscript)) + + template = _resolve_domain(manuscript.domain) + summary_parts = [] + if template is not None: + summary_parts.append(template["intro"]) + summary_parts.append(template["checks"]) + else: + summary_parts.append("Review generated with the generic cross-domain template.") + + report = ReviewReport( + title=manuscript.title, + domain=manuscript.domain, + issues=issues, + summary=" ".join(summary_parts), + ) + return report + + +def peer_review_score(report: ReviewReport) -> int: + """Return the 0-100 score of a :class:`ReviewReport`. + + The score starts at 100 and deducts points weighted by issue severity + (errors weigh more than warnings, which weigh more than informational + notes). + + Parameters + ---------- + report: + The review report to score. + + Returns + ------- + int + Score between 0 and 100. + """ + return report.score \ No newline at end of file diff --git a/tests/test_scibase.py b/tests/test_scibase.py new file mode 100644 index 00000000..be5a7921 --- /dev/null +++ b/tests/test_scibase.py @@ -0,0 +1,374 @@ +"""Tests for the AI-Powered Research Assistant Suite.""" + +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from scibase.gap_finder import GapFinder, Paper, ResearchOpportunity, extract_questions +from scibase.reproducibility import Project, ReproducibilityChecker +from scibase.review import ( + Manuscript, + ReviewCategory, + ReviewReport, + generate_peer_review, + peer_review_score, +) + + +# --------------------------------------------------------------------------- +# 1. Auto peer review reports +# --------------------------------------------------------------------------- + + +class TestPeerReview: + def test_clean_manuscript_scores_100(self): + manuscript = Manuscript( + title="A complete study", + abstract=( + "We measured the growth rate of 120 cell cultures over 30 days. " + "The mean growth rate was 0.42 (p = 0.023). We found a strong " + "correlation between growth and temperature (r = 0.81). The " + "response rate of the follow-up survey was 73 percent." + ), + body=( + "Methods: samples were collected and analyzed. Results confirm " + "the hypothesized effect in a carefully controlled setting. " + "The effect size remained consistent across repeated trials." + ), + claims=[ + ("Growth rate is temperature dependent.", "Measured r = 0.81 across 120 cultures."), + ], + citations=["Smith et al., 2021", "Jones & Doe, 2022"], + ) + report = generate_peer_review(manuscript) + assert report.title == "A complete study" + assert report.score == 100 + assert report.issues == [] + + def test_missing_citations_flagged(self): + manuscript = Manuscript( + title="No citations", + abstract="We measured the growth rate of 120 cultures over 30 days.", + body="The growth rate is shown to be temperature dependent.", + ) + report = generate_peer_review(manuscript) + assert any(i.category == ReviewCategory.CITATION for i in report.issues) + assert any(i.severity == "error" for i in report.issues if i.category == ReviewCategory.CITATION) + + def test_unsupported_claim_flagged(self): + manuscript = Manuscript( + title="Unsupported claim", + abstract="We measured the growth rate of 120 cultures over 30 days.", + claims=[("This treatment cures the disease.", None)], + citations=["Smith et al., 2021"], + ) + report = generate_peer_review(manuscript) + claims_issues = [i for i in report.issues if i.category == ReviewCategory.CLAIMS] + assert any("cures the disease" in i.message for i in claims_issues) + assert all(i.severity == "error" for i in claims_issues) + + def test_statistical_red_flag_unreported_pvalue(self): + manuscript = Manuscript( + title="Statistical red flag", + abstract="A significant p-value was observed.", + body="The p-value was below the threshold.", + citations=["Smith et al., 2021"], + ) + report = generate_peer_review(manuscript) + stats = [i for i in report.issues if i.category == ReviewCategory.STATISTICAL] + assert any("p-value" in i.message for i in stats) + assert any(i.severity == "error" for i in stats) + + def test_domain_specific_template(self): + manuscript = Manuscript( + title="Clinical trial", + abstract="We measured the growth rate of 120 cultures over 30 days.", + body="", + domain="clinical-trials", + citations=["Smith et al., 2021"], + ) + report = generate_peer_review(manuscript) + assert "clinical trial reporting standards" in report.summary + + def test_unknown_domain_falls_back_to_generic(self): + manuscript = Manuscript( + title="Weird domain", + abstract="We measured the growth rate of 120 cultures over 30 days.", + domain="alchemy", + ) + report = generate_peer_review(manuscript) + assert "generic cross-domain template" in report.summary + + def test_score_deductions(self): + errors = [Manuscript( + title="x", + abstract="t", + body="", + )] + clean = Manuscript( + title="clean", + abstract="We measured the growth rate of 120 cultures over 30 days.", + body="", + citations=["a", "b"], + ) + bad_score = peer_review_score(generate_peer_review(errors[0])) + good_score = peer_review_score(generate_peer_review(clean)) + assert bad_score < good_score + assert 0 <= bad_score <= 100 + + def test_report_is_review_report_type(self): + report = generate_peer_review(Manuscript(title="t", abstract="a")) + assert isinstance(report, ReviewReport) + + +# --------------------------------------------------------------------------- +# 2. Reproducibility checker +# --------------------------------------------------------------------------- + + +class TestReproducibility: + def _write_pipeline(self, tmp_path): + (tmp_path / "analysis.py").write_text("print('done')\n") + data = tmp_path / "data" + data.mkdir() + (data / "raw.csv").write_text("x,y\n1,2\n") + tests = tmp_path / "tests" + tests.mkdir() + (tests / "test_analysis.py").write_text("def test_x(): pass\n") + + def test_complete_project_scores_full(self, tmp_path): + self._write_pipeline(tmp_path) + (tmp_path / "results.txt").write_text("accuracy=0.94\n") + + project = Project( + name="complete", + root=tmp_path, + reported_results={"results.txt": "accuracy=0.94"}, + dependencies={"numpy": "1.26.4", "pandas": "2.2.2"}, + ) + report = ReproducibilityChecker().check(project) + assert report.score == 1.0 + assert report.reproducible + assert set(report.checks_passed) == { + "pipeline", + "raw-data", + "tests", + "output-consistency", + "dependencies", + "determinism", + } + + def test_missing_pipeline_and_data_flagged(self, tmp_path): + project = Project(name="empty", root=tmp_path) + report = ReproducibilityChecker().check(project) + codes = {i.code for i in report.issues} + assert {"no-pipeline", "no-raw-data", "no-tests"} <= codes + assert report.score == 0.4 + assert not report.reproducible + + def test_output_mismatch_flagged(self, tmp_path): + self._write_pipeline(tmp_path) + (tmp_path / "results.txt").write_text("accuracy=0.91\n") + + project = Project( + name="mismatch", + root=tmp_path, + reported_results={"results.txt": "accuracy=0.94"}, + ) + report = ReproducibilityChecker().check(project) + assert any(i.code == "output-mismatch" for i in report.issues) + assert "output-consistency" not in report.checks_passed + + def test_missing_reported_output_flagged(self, tmp_path): + self._write_pipeline(tmp_path) + + project = Project( + name="missing-output", + root=tmp_path, + reported_results={"results.txt": "accuracy=0.94"}, + ) + report = ReproducibilityChecker().check(project) + assert any(i.code == "missing-output" for i in report.issues) + + def test_unpinned_dependency_flagged(self, tmp_path): + self._write_pipeline(tmp_path) + + project = Project( + name="unpinned", + root=tmp_path, + dependencies={"numpy": None}, + ) + report = ReproducibilityChecker().check(project) + assert any(i.code == "unpinned-dependency" for i in report.issues) + assert "dependencies" not in report.checks_passed + + def test_non_determinism_detected(self, tmp_path): + self._write_pipeline(tmp_path) + + runs = tmp_path / "reproducibility" / "runs" + (runs / "run1").mkdir(parents=True) + (runs / "run2").mkdir(parents=True) + (runs / "run1" / "output.txt").write_text("same\n") + (runs / "run2" / "output.txt").write_text("different\n") + + project = Project(name="non-deterministic", root=tmp_path) + report = ReproducibilityChecker().check(project) + assert any(i.code == "non-deterministic" for i in report.issues) + assert "determinism" not in report.checks_passed + + def test_deterministic_runs_pass(self, tmp_path): + self._write_pipeline(tmp_path) + + runs = tmp_path / "reproducibility" / "runs" + (runs / "run1").mkdir(parents=True) + (runs / "run2").mkdir(parents=True) + (runs / "run1" / "output.txt").write_text("same\n") + (runs / "run2" / "output.txt").write_text("same\n") + + project = Project(name="deterministic", root=tmp_path) + report = ReproducibilityChecker().check(project) + assert "determinism" in report.checks_passed + assert not any(i.code == "non-deterministic" for i in report.issues) + + +# --------------------------------------------------------------------------- +# 3. Research gap finder +# --------------------------------------------------------------------------- + + +def _corpus() -> list[Paper]: + return [ + Paper( + id="p1", + title="CRISPR screens in single-cell RNA-seq", + abstract="", + topics=["CRISPR", "Alzheimer's", "single-cell RNA-seq"], + citations=40, + open_questions=["The role of glia remains unclear.", "Future work should map cell states."], + ), + Paper( + id="p2", + title="Single-cell atlas of Alzheimer's", + abstract="", + topics=["Alzheimer's", "single-cell RNA-seq"], + citations=30, + ), + Paper( + id="p3", + title="CRISPR for gene therapy", + abstract="", + topics=["CRISPR", "gene therapy"], + citations=25, + ), + Paper( + id="p4", + title="CRISPR editing in neurons", + abstract="", + topics=["CRISPR", "Alzheimer's"], + citations=20, + ), + Paper( + id="p5", + title="Single-cell methods review", + abstract="", + topics=["single-cell RNA-seq"], + citations=10, + ), + Paper( + id="p6", + title="Gene therapy trial", + abstract="", + topics=["gene therapy"], + citations=3, + ), + Paper( + id="p7", + title="Alzheimer's pathology", + abstract="", + topics=["Alzheimer's"], + citations=15, + ), + ] + + +class TestGapFinder: + def test_topic_popularity(self): + finder = GapFinder(_corpus()) + popularity = finder.topic_popularity() + assert popularity["CRISPR"] == 3 + assert popularity["Alzheimer's"] == 4 + assert popularity["single-cell RNA-seq"] == 3 + assert popularity["gene therapy"] == 2 + + def test_under_studied_intersection_found(self): + finder = GapFinder(_corpus()) + opportunities = finder.under_studied_intersections() + combos = {o.topics for o in opportunities} + # CRISPR + single-cell RNA-seq never co-occur in the corpus + assert frozenset({"CRISPR", "single-cell RNA-seq"}) in combos + # but gene therapy is not active enough to qualify + assert all("gene therapy" not in o.topics for o in opportunities) + assert opportunities == sorted(opportunities, key=lambda o: o.score, reverse=True) + + def test_saturated_intersection_excluded(self): + papers = [ + Paper(id="a", title="a", topics=["X", "Y"], citations=10), + Paper(id="b", title="b", topics=["X", "Y"], citations=10), + Paper(id="c", title="c", topics=["X"], citations=10), + Paper(id="d", title="d", topics=["Y"], citations=10), + ] + finder = GapFinder(papers) + opportunities = finder.under_studied_intersections() + # every paper covering X or Y already covers both -> no gap + assert all(frozenset({"X", "Y"}) != o.topics for o in opportunities) + + def test_unresolved_questions(self): + finder = GapFinder(_corpus()) + questions = finder.unresolved_questions(min_citations=5) + assert any(q.startswith("The role of glia") for q, _, _ in questions) + assert all(total >= 5 for _, _, total in questions) + # sorted by total citations descending + assert questions == sorted(questions, key=lambda item: item[2], reverse=True) + + def test_research_opportunities_feed_personalized(self): + finder = GapFinder(_corpus()) + feed = finder.research_opportunities_feed( + interests=["single-cell RNA-seq"], + project_history=["CRISPR"], + ) + assert feed + assert all(o.relevance >= 1 for o in feed) + assert all(o.topics & {"single-cell RNA-seq", "CRISPR"} for o in feed) + assert feed == sorted(feed, key=lambda o: o.score, reverse=True) + + def test_feed_excludes_irrelevant_gaps(self): + finder = GapFinder(_corpus()) + feed = finder.research_opportunities_feed( + interests=["gene therapy"], + project_history=[], + ) + assert feed == [] + + def test_extract_questions(self): + text = ( + "The mechanism remains unclear. " + "Future work should address this. " + "Results were conclusive." + ) + questions = extract_questions(text) + assert len(questions) == 2 + assert "remains unclear" in questions[0] + + def test_opportunity_score(self): + opportunity = ResearchOpportunity( + topics=frozenset({"CRISPR", "single-cell RNA-seq"}), + popularity={"CRISPR": 3, "single-cell RNA-seq": 3}, + combination_count=0, + rationale="never co-occur", + relevance=2, + ) + # (3 + 3) * (1 + 2) / (1 + 0) = 18 + assert opportunity.score == 18.0 \ No newline at end of file