diff --git a/pyproject.toml b/pyproject.toml index 761397b..5efcb92 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,6 +43,7 @@ dependencies = [ "accelerate", "bitsandbytes", "biomcp-python", + "biodb @ git+https://github.com/bschilder/bioDB", ] [project.optional-dependencies] diff --git a/synthlab/download_snomed.py b/synthlab/download_snomed.py index c8acf45..bde3c79 100644 --- a/synthlab/download_snomed.py +++ b/synthlab/download_snomed.py @@ -1,324 +1,95 @@ -#!/usr/bin/env python3 -""" -Download SNOMED CT vocabulary data for SynthLab. - -The CONCEPT.csv file is hosted on GitHub releases and downloaded on first use. -For private repos, authentication is handled via: -1. GitHub CLI (gh) if available and authenticated -2. GITHUB_TOKEN environment variable -3. GH_TOKEN environment variable -""" - -import gzip -import os -import shutil -import subprocess -import urllib.request -from pathlib import Path - -# GitHub repository and release info -GITHUB_REPO = "bschilder/synthlab" -GITHUB_RELEASE_TAG = "vocab-v1" -GITHUB_ASSET_NAME = "CONCEPT.csv.gz" - -# Direct URL (works for public repos) -SNOMED_RELEASE_URL = f"https://github.com/{GITHUB_REPO}/releases/download/{GITHUB_RELEASE_TAG}/{GITHUB_ASSET_NAME}" - -# Default cache location -DEFAULT_SNOMED_DATA_DIR = Path.home() / ".cache" / "synthlab" / "snomed_data" - - -def get_snomed_data_dir() -> Path: - """Get the SNOMED data directory, creating if needed.""" - DEFAULT_SNOMED_DATA_DIR.mkdir(parents=True, exist_ok=True) - return DEFAULT_SNOMED_DATA_DIR - - -def get_concept_csv_path() -> Path: - """Get the path to CONCEPT.csv, downloading if needed.""" - data_dir = get_snomed_data_dir() - concept_path = data_dir / "CONCEPT.csv" - - if not concept_path.exists(): - download_snomed_vocabulary(verbose=True) - - return concept_path - - -def _get_github_token() -> str | None: - """Get GitHub token from environment variables.""" - return os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN") - - -def _find_gh_cli() -> str | None: - """Find the gh CLI executable path.""" - import sys - - # Check common locations - candidates = [ - "gh", # In PATH - Path(sys.executable).parent / "gh", # Same dir as Python (conda env) - ] - - for candidate in candidates: - try: - result = subprocess.run( - [str(candidate), "--version"], - capture_output=True, - timeout=5, - ) - if result.returncode == 0: - return str(candidate) - except (subprocess.SubprocessError, FileNotFoundError, OSError): - continue - - return None - - -def _gh_cli_available() -> bool: - """Check if GitHub CLI is available and authenticated.""" - gh_path = _find_gh_cli() - if not gh_path: - return False - - try: - result = subprocess.run( - [gh_path, "auth", "status"], - capture_output=True, - text=True, - timeout=10, - ) - return result.returncode == 0 - except (subprocess.SubprocessError, FileNotFoundError): - return False - - -def _download_with_gh_cli(output_path: Path, verbose: bool = True) -> bool: - """ - Download release asset using GitHub CLI. - - Returns True if successful, False otherwise. - """ - gh_path = _find_gh_cli() - if not gh_path: - return False - - try: - if verbose: - print(" Using GitHub CLI for authenticated download...") - - result = subprocess.run( - [ - gh_path, "release", "download", GITHUB_RELEASE_TAG, - "--repo", GITHUB_REPO, - "--pattern", GITHUB_ASSET_NAME, - "--dir", str(output_path.parent), - "--clobber", - ], - capture_output=True, - text=True, - timeout=300, # 5 minute timeout for large files - ) - - if result.returncode == 0: - return True - - if verbose: - print(f" gh CLI failed: {result.stderr.strip()}") - return False - - except (subprocess.SubprocessError, FileNotFoundError) as e: - if verbose: - print(f" gh CLI error: {e}") - return False +"""Back-compat shim — the SNOMED downloader now lives in :mod:`biodb.snomed`. +The full implementation (with tqdm progress, the 3-strategy auth +flow, and the OHDSI ``CONCEPT.csv`` loader) was relocated to bioDB on +2026-05-18. The GitHub Release asset also moved — same bytes, same +SHA-256, new home at ``bschilder/bioDB`` (release ``vocab-v1``). -def _download_with_token(url: str, output_path: Path, token: str, verbose: bool = True) -> bool: - """ - Download using GitHub token authentication. - - Returns True if successful, False otherwise. - """ - try: - if verbose: - print(" Using token authentication...") +This module re-exports the public names so any existing +``from synthlab.download_snomed import ...`` keeps working. New code +should import directly from :mod:`biodb.snomed`. +""" - # For GitHub release assets, we need to use the API - # First get the asset URL, then download with token - api_url = f"https://api.github.com/repos/{GITHUB_REPO}/releases/tags/{GITHUB_RELEASE_TAG}" +from __future__ import annotations - request = urllib.request.Request(api_url) - request.add_header("Authorization", f"token {token}") - request.add_header("Accept", "application/vnd.github.v3+json") +import warnings - with urllib.request.urlopen(request, timeout=30) as response: - import json - release_data = json.loads(response.read().decode()) +from biodb.snomed import ( + CACHE_DIR as _BIODB_CACHE_DIR, +) +from biodb.snomed import ( + GITHUB_ASSET_NAME, + GITHUB_RELEASE_TAG, + GITHUB_REPO, + SNOMED_RELEASE_URL, + download_concept_csv as _download_concept_csv, +) +from biodb.snomed import ( + get_concept_csv_path as _get_concept_csv_path, +) +from biodb.snomed import ( + get_snomed_data_dir as _get_snomed_data_dir, +) +from biodb.snomed import ( + is_available as _is_available, +) - # Find the asset - asset_url = None - for asset in release_data.get("assets", []): - if asset["name"] == GITHUB_ASSET_NAME: - asset_url = asset["url"] - break +# Back-compat alias for the cache directory constant. +DEFAULT_SNOMED_DATA_DIR = _BIODB_CACHE_DIR - if not asset_url: - if verbose: - print(f" Asset {GITHUB_ASSET_NAME} not found in release") - return False +# Re-export under the original synthlab names. The original module had +# slightly different function signatures (``output_dir`` first, plus +# ``url=`` and ``verbose=`` kwargs) — wrap to preserve those callers. - # Download the asset - request = urllib.request.Request(asset_url) - request.add_header("Authorization", f"token {token}") - request.add_header("Accept", "application/octet-stream") - with urllib.request.urlopen(request, timeout=300) as response: - with open(output_path, 'wb') as f: - shutil.copyfileobj(response, f) +def get_snomed_data_dir(): # type: ignore[no-redef] + """Re-export of :func:`biodb.snomed.get_snomed_data_dir`.""" + return _get_snomed_data_dir() - return True - except Exception as e: - if verbose: - print(f" Token auth failed: {e}") - return False +def get_concept_csv_path(): # type: ignore[no-redef] + """Re-export of :func:`biodb.snomed.get_concept_csv_path`.""" + return _get_concept_csv_path() -def _download_public(url: str, output_path: Path, verbose: bool = True) -> bool: - """ - Download from public URL (no authentication). - - Returns True if successful, False otherwise. - """ - try: - if verbose: - print(" Attempting public download...") - urllib.request.urlretrieve(url, output_path) - return True - except urllib.error.HTTPError as e: - if verbose: - print(f" Public download failed: HTTP {e.code}") - return False - except Exception as e: - if verbose: - print(f" Public download failed: {e}") - return False +def is_snomed_available() -> bool: + """Back-compat name for :func:`biodb.snomed.is_available`.""" + return _is_available() def download_snomed_vocabulary( - output_dir: Path | str | None = None, + output_dir=None, url: str = SNOMED_RELEASE_URL, verbose: bool = True, force: bool = False, -) -> Path: +): + """Back-compat wrapper for :func:`biodb.snomed.download_concept_csv`. + + The ``url`` argument is accepted for ABI compatibility but is now + ignored — :mod:`biodb.snomed` always uses the release URL on the + bioDB repo. If you were overriding ``url`` to point at a private + mirror, set ``GITHUB_TOKEN`` instead and bioDB will use the token + auth flow against the same release tag. """ - Download SNOMED CT CONCEPT.csv from GitHub releases. - - For private repositories, authentication is handled automatically via: - 1. GitHub CLI (gh) if available and authenticated - 2. GITHUB_TOKEN or GH_TOKEN environment variable - - Args: - output_dir: Directory to save CONCEPT.csv. Defaults to ~/.cache/synthlab/snomed_data/ - url: URL to download from (defaults to GitHub releases) - verbose: Print progress messages - force: Re-download even if file exists - - Returns: - Path to the downloaded CONCEPT.csv file - """ - if output_dir is None: - output_dir = get_snomed_data_dir() - else: - output_dir = Path(output_dir) - output_dir.mkdir(parents=True, exist_ok=True) - - concept_path = output_dir / "CONCEPT.csv" - compressed_path = output_dir / "CONCEPT.csv.gz" - - # Check if already downloaded - if concept_path.exists(): - if verbose: - print(f"SNOMED CONCEPT.csv already exists at {concept_path}") - if force: - print("Skipping download (delete the file to re-download).") - return concept_path - - if verbose: - print(f"Downloading SNOMED vocabulary from GitHub releases...") - print(f" Repository: {GITHUB_REPO}") - print(f" Release: {GITHUB_RELEASE_TAG}") - print(f" Destination: {concept_path}") - - try: - # Try download methods in order of preference - download_success = False - - # 1. Try GitHub CLI (best for private repos) - if _gh_cli_available(): - download_success = _download_with_gh_cli(compressed_path, verbose) - - # 2. Try token authentication - if not download_success: - token = _get_github_token() - if token: - download_success = _download_with_token(url, compressed_path, token, verbose) - - # 3. Try public download (works if repo is public) - if not download_success: - download_success = _download_public(url, compressed_path, verbose) - - if not download_success: - raise RuntimeError( - "Failed to download SNOMED vocabulary.\n\n" - "For private repositories, ensure one of:\n" - " 1. GitHub CLI is installed and authenticated: gh auth login\n" - " 2. GITHUB_TOKEN environment variable is set\n" - " 3. GH_TOKEN environment variable is set\n\n" - f"Repository: {GITHUB_REPO}\n" - f"Release: {GITHUB_RELEASE_TAG}" - ) - - if verbose: - size_mb = compressed_path.stat().st_size / (1024 * 1024) - print(f" Downloaded ({size_mb:.1f} MB)") - - # Decompress - if verbose: - print(" Decompressing...", end=" ", flush=True) - - with gzip.open(compressed_path, 'rb') as f_in: - with open(concept_path, 'wb') as f_out: - shutil.copyfileobj(f_in, f_out) - - if verbose: - size_mb = concept_path.stat().st_size / (1024 * 1024) - print(f"done ({size_mb:.1f} MB)") - - # Remove compressed file - compressed_path.unlink() - - if verbose: - print(f" SNOMED CONCEPT.csv saved to {concept_path}") - - return concept_path - - except Exception as e: - # Clean up partial downloads - if compressed_path.exists(): - compressed_path.unlink() - if concept_path.exists(): - concept_path.unlink() - raise - - -def is_snomed_available() -> bool: - """Check if SNOMED CONCEPT.csv is available locally.""" - concept_path = get_snomed_data_dir() / "CONCEPT.csv" - return concept_path.exists() - - -if __name__ == "__main__": - print("SNOMED Vocabulary Downloader") - print("=" * 60) - download_snomed_vocabulary(verbose=True) + if url != SNOMED_RELEASE_URL: + warnings.warn( + f"synthlab.download_snomed.download_snomed_vocabulary(url=...) is " + f"ignored — biodb.snomed always uses {SNOMED_RELEASE_URL}. " + f"Set GITHUB_TOKEN / GH_TOKEN for private-mirror access.", + DeprecationWarning, + stacklevel=2, + ) + return _download_concept_csv(output_dir=output_dir, force=force, progress=verbose) + + +__all__ = [ + "DEFAULT_SNOMED_DATA_DIR", + "GITHUB_ASSET_NAME", + "GITHUB_RELEASE_TAG", + "GITHUB_REPO", + "SNOMED_RELEASE_URL", + "download_snomed_vocabulary", + "get_concept_csv_path", + "get_snomed_data_dir", + "is_snomed_available", +]