Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion .gitignore
Original file line number Diff line number Diff line change
@@ -1,2 +1,8 @@
__pycache__
_build
_build
build/
*.egg-info/

# Legacy in-package taxonomy cache written by bkbit < this release; the cache
# now lives in a per-user cache dir (see bkbit/utils/ncbi_taxonomy_cache.py).
bkbit/utils/ncbi_taxonomy/
32 changes: 30 additions & 2 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ The package provides a CLI tool `bkbit` with multiple subcommands:
- `bkbit schema2model` - Convert spreadsheets to LinkML YAML models
- `bkbit yaml2cvs` - Convert YAML to CSV
- `bkbit linkml-trimmer` - Trim LinkML models
- `bkbit download-ncbi-taxonomy` - Download NCBI taxonomy data
- `bkbit download-ncbi-taxonomy` - Optionally pre-download the full NCBI taxonomy (see below)

## Architecture

Expand All @@ -73,7 +73,9 @@ The package provides a CLI tool `bkbit` with multiple subcommands:
- `add_dunderMethods_genomeAnnotation.py` - Adds __eq__, __ne__, __hash__ to GeneAnnotation

**`bkbit/utils/`** - Shared utilities:
- `get_ncbi_taxonomy.py` - NCBI taxonomy download
- `ncbi_taxonomy_cache.py` - NCBI taxonomy name lookups (see NCBI Taxonomy Data below)
- `ncbi_taxonomy_data/` - Bundled taxonomy subset + the script that rebuilds it
- `get_ncbi_taxonomy.py` - Thin CLI wrapper for pre-fetching the full taxonomy
- `nimp_api_endpoints.py` - Specimen Portal API endpoints
- `setup_logger.py` - Logging configuration

Expand All @@ -82,5 +84,31 @@ The package provides a CLI tool `bkbit` with multiple subcommands:
- **Schemasheets** - Spreadsheet to LinkML conversion
- **Click** - CLI framework

### NCBI Taxonomy Data

`gff2jsonld` resolves organism names through `bkbit/utils/ncbi_taxonomy_cache.py`, which
serves lookups from two layers and requires no setup after `pip install`:

1. **Bundled subset** (`bkbit/utils/ncbi_taxonomy_data/taxonomy_subset.json.gz`, ~575KB) -
every taxon with a GenBank common name (~30k), which covers all BICAN organisms.
Offline, loads in milliseconds.
2. **Full NCBI taxonomy** - downloaded on demand only if a lookup misses the subset, and
cached in a per-user cache directory (never in `site-packages`).

Rules to preserve when touching this code:
- **Never load taxonomy data at import time.** The previous version read the maps in the
`Gff3` class body, so a missing cache broke every `bkbit` subcommand, including `--help`.
- **Never write runtime data into the installed package.** `site-packages` may be read-only
and is wiped on upgrade.

To regenerate the bundled subset for a newer taxonomy dump:
```bash
bkbit download-ncbi-taxonomy
python -m bkbit.utils.ncbi_taxonomy_data.build_subset # commit the resulting .json.gz
```

### Environment Variables
- `jwt_token` - Specimen Portal Personal API Token (required for specimen2jsonld)
- `BKBIT_DATA_DIR` - Overrides the cache directory for the downloaded full NCBI taxonomy
- `BKBIT_NO_DOWNLOAD` - If set, a taxonomy lookup that misses the bundled subset raises
instead of downloading (useful in CI/air-gapped runs)
1 change: 0 additions & 1 deletion bkbit/data_translators/HMBA_annotation_translator.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
import itertools
import click
import pkg_resources
from bkbit.models import bke_taxonomy
import json
import hashlib
Expand Down
37 changes: 17 additions & 20 deletions bkbit/data_translators/genome_annotation_translator.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,12 +41,11 @@
import gzip
from tqdm import tqdm
import click
import pkg_resources
from linkml_runtime.dumpers import json_dumper
from rdflib import Graph
from bkbit.models import genome_annotation as ga
from bkbit.utils.setup_logger import setup_logger
from bkbit.utils.load_json import load_json
from bkbit.utils import ncbi_taxonomy_cache
from bkbit.utils.generate_bkbit_id import generate_object_id
from bkbit.utils.serialize_to_ttl import convert_jsonld_to_ttl

Expand All @@ -72,10 +71,6 @@
)
DEFAULT_FEATURE_FILTER = ("gene", "pseudogene", "ncRNA_gene")
DEFAULT_HASH = ("MD5",)
TAXON_DIR_PATH = "../utils/ncbi_taxonomy/"
SCIENTIFIC_NAME_TO_TAXONID_PATH = pkg_resources.resource_filename(__name__, TAXON_DIR_PATH + "scientific_name_to_taxid.json")
TAXON_SCIENTIFIC_NAME_PATH = pkg_resources.resource_filename(__name__, TAXON_DIR_PATH + "taxid_to_scientific_name.json")
TAXON_COMMON_NAME_PATH = pkg_resources.resource_filename(__name__, TAXON_DIR_PATH + "taxid_to_common_name.json")
INCOMPATABLE_EXTENSION = "The provided content URL is not supported. Please provide a valid URL with '.gff.gz' extension."
class Gff3:
"""
Expand Down Expand Up @@ -134,18 +129,10 @@
serialize_to_jsonld(exclude_none=True, exclude_unset=False):
Serializes the object and either writes it to the specified output file or prints it to the CLI.
"""
scientific_name_to_taxonid = None
taxon_scientific_name = None
taxon_common_name = None

# Load taxonomy data at class definition
try:
scientific_name_to_taxonid = load_json(SCIENTIFIC_NAME_TO_TAXONID_PATH)
taxon_scientific_name = load_json(TAXON_SCIENTIFIC_NAME_PATH)
taxon_common_name = load_json(TAXON_COMMON_NAME_PATH)
except FileNotFoundError as e:
#logging.critical("NCBI Taxonomy not downloaded. Run 'bkbit download-ncbi-taxonomy' first.")
raise RuntimeError("NCBI Taxonomy not downloaded. Run 'bkbit download-ncbi-taxonomy' first.") from e
# Taxonomy names are resolved lazily, one taxon at a time, by
# bkbit.utils.ncbi_taxonomy_cache. Nothing is loaded or downloaded until a
# lookup actually happens, so importing this module is always side-effect
# free.

def __init__(
self,
Expand Down Expand Up @@ -299,7 +286,7 @@
)

scientific_name = ensembl_match.group(3)
taxonid = self.scientific_name_to_taxonid.get(
taxonid = ncbi_taxonomy_cache.lookup_taxid(
scientific_name.replace("_", " ")
)
return {
Expand Down Expand Up @@ -377,8 +364,18 @@

Returns:
ga.OrganismTaxon: The generated organism taxon object.

Raises:
ValueError: If the taxon ID is not present in the NCBI taxonomy.
"""
attributes = {"full_name": cls.taxon_scientific_name[taxon_id], "name": cls.taxon_common_name[taxon_id], "iri": PREFIX_MAP[TAXON_PREFIX] + taxon_id, "xref": [TAXON_PREFIX + taxon_id]}
scientific_name = ncbi_taxonomy_cache.lookup_scientific_name(taxon_id)
if scientific_name is None:
raise ValueError(
f"Taxon ID '{taxon_id}' was not found in the NCBI taxonomy."
)
# Not every taxon has a GenBank common name; fall back to the scientific name.
common_name = ncbi_taxonomy_cache.lookup_common_name(taxon_id) or scientific_name
attributes = {"full_name": scientific_name, "name": common_name, "iri": PREFIX_MAP[TAXON_PREFIX] + taxon_id, "xref": [TAXON_PREFIX + taxon_id]}
attributes["id"] = generate_object_id(attributes)
return ga.OrganismTaxon(**attributes)

Expand Down Expand Up @@ -587,7 +584,7 @@
biotype = self._get_attribute(attributes, "biotype", curr_line_num)

attributes = {"source_id": stable_id, "symbol": name, "name": name, "description": description, "molecular_type": biotype, "referenced_in": self.genome_annotation.id, "in_taxon": [self.organism_taxon.id], "in_taxon_label": self.organism_taxon.full_name, "xref": [ENSEMBL_GENE_ID_PREFIX + stable_id]}
#! add a try/catch incase the hash returns an error and log it

Check failure on line 587 in bkbit/data_translators/genome_annotation_translator.py

View workflow job for this annotation

GitHub Actions / Check for spelling errors

incase ==> in case
attributes["id"] = generate_object_id(attributes)
gene_annotation = ga.GeneAnnotation(**attributes)

Expand Down Expand Up @@ -659,7 +656,7 @@
)

attributes = {"source_id": stable_id, "symbol": name, "name": name, "description": description, "molecular_type": biotype, "referenced_in": self.genome_annotation.id, "in_taxon": [self.organism_taxon.id], "in_taxon_label": self.organism_taxon.full_name, "synonym": synonyms, "xref": [NCBI_GENE_ID_PREFIX + stable_id]}
#! add a try/catch incase the hash returns an error and log it

Check failure on line 659 in bkbit/data_translators/genome_annotation_translator.py

View workflow job for this annotation

GitHub Actions / Check for spelling errors

incase ==> in case
attributes["id"] = generate_object_id(attributes)
gene_annotation = ga.GeneAnnotation(**attributes)

Expand Down
4 changes: 2 additions & 2 deletions bkbit/model_editors/add_dunderMethods_genomeAnnotation.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import re
import pkg_resources
from pathlib import Path
# Read the file
genome_annotation_model = pkg_resources.resource_filename(__name__, "../models/genome_annotation.py")
genome_annotation_model = Path(__file__).parent.parent / "models" / "genome_annotation.py"

#file_path = "bkbit/models/genome_annotation.py"
with open(genome_annotation_model, "r") as file:
Expand Down
219 changes: 52 additions & 167 deletions bkbit/utils/get_ncbi_taxonomy.py
Original file line number Diff line number Diff line change
@@ -1,187 +1,72 @@
"""
This script downloads a zip file containing taxonomic data from a given URL, extracts and processes
the content of the 'names.dmp' file in memory, and saves the parsed data into JSON files. The script
includes three main functions:
CLI for pre-fetching the full NCBI taxonomy cache.

1. download_and_extract_zip_in_memory(url):
Downloads a zip file from the given URL and extracts the content of the 'names.dmp' file in memory.
bkbit does not require this step: a taxonomy subset covering every organism with
a GenBank common name ships inside the wheel, and the full taxonomy is downloaded
automatically the first time a lookup falls outside that subset. This command
exists for the cases where you want the download to happen up front rather than
mid-pipeline - air-gapped runs, container images, CI, or reproducible builds.

2. parse_dmp_content(dmp_content):
Parses the content of a DMP file and extracts taxonomic information into dictionaries.

3. process_and_save_taxdmp_in_memory(url, output_dir):
Downloads and processes the taxdump file from the given URL, and saves the parsed data into
separate JSON files in the specified output directory.
All of the download and cache logic lives in
:mod:`bkbit.utils.ncbi_taxonomy_cache`; this module is a thin wrapper around it
and re-exports the previous function names for backwards compatibility.

Usage:
The script can be executed as a standalone program. Modify the URL and output directory as needed.
bkbit download-ncbi-taxonomy [--reload] [--data-dir PATH]
"""

import json
import zipfile
import io
import os
import requests
import pkg_resources
import click

NCBI_TAXON_URL = "https://ftp.ncbi.nih.gov/pub/taxonomy/taxdmp.zip"
OUTPUT_DIR_NAME = "ncbi_taxonomy"
OUTPUT_DIR_PATH = pkg_resources.resource_filename(__name__, OUTPUT_DIR_NAME)
SCIENTIFIC_NAME_TO_TAXONID_PATH = pkg_resources.resource_filename(__name__, "ncbi_taxonomy/scientific_name_to_taxid.json")
TAXON_SCIENTIFIC_NAME_PATH = pkg_resources.resource_filename(__name__, "ncbi_taxonomy/taxid_to_scientific_name.json")
TAXON_COMMON_NAME_PATH = pkg_resources.resource_filename(__name__, "ncbi_taxonomy/taxid_to_common_name.json")



def download_and_extract_zip_in_memory(url):
from bkbit.utils.ncbi_taxonomy_cache import (
NCBI_TAXON_URL,
build_full_cache,
download_and_extract_zip_in_memory,
ensure_full_cache,
parse_dmp_content,
)

__all__ = [
"NCBI_TAXON_URL",
"build_full_cache",
"download_and_extract_zip_in_memory",
"download_ncbi_taxonomy",
"parse_dmp_content",
"process_and_save_taxdmp_in_memory",
]


def process_and_save_taxdmp_in_memory(url=NCBI_TAXON_URL, output_dir=None):
"""
Downloads a zip file from the given URL and extracts the content of the 'names.dmp' file in memory.

Args:
url (str): The URL of the zip file to download.

Returns:
str: The content of the 'names.dmp' file as a string.

Raises:
requests.exceptions.HTTPError: If the file download fails with a non-200 status code.
"""
# Download the file
response = requests.get(url, timeout=30)
if response.status_code == 200:
# Unzip the file in memory
with zipfile.ZipFile(io.BytesIO(response.content)) as z:
# Extract names.dmp file content into memory
with z.open("names.dmp") as names_dmp_file:
names_dmp_content = names_dmp_file.read().decode("utf-8")
return names_dmp_content
else:
raise requests.exceptions.HTTPError(
f"Failed to download file, status code: {response.status_code}"
)

Downloads and processes the taxdump file, saving the parsed data as JSON.

def parse_dmp_content(dmp_content):
"""
Parses the content of a DMP file and extracts taxonomic information.
Deprecated alias for :func:`bkbit.utils.ncbi_taxonomy_cache.build_full_cache`.

Args:
dmp_content (str): The content of the DMP file.

Returns:
tuple: A tuple containing three dictionaries:
- taxid_to_scientific_name: A dictionary mapping taxonomic IDs to scientific names.
- taxid_to_common_name: A dictionary mapping taxonomic IDs to common names.
- scientific_name_to_taxid: A dictionary mapping scientific names to taxonomic IDs.
url: The URL of the taxdump file to download and process.
output_dir: Destination directory. Defaults to the bkbit cache directory.
"""
taxid_to_scientific_name = {}
taxid_to_common_name = {}
scientific_name_to_taxid = {}

for line in dmp_content.strip().split("\n"):
# Split the line by the delimiter '|'
parts = line.strip().split("|")

# Remove leading and trailing whitespace from each part
parts = [part.strip() for part in parts]
# Taxonomy names file (names.dmp):
# tax_id-- the id of node associated with this name
# name_txt-- name itself
# unique name-- the unique variant of this name if name not unique
# name class-- (synonym, common name, ...)
taxid = parts[0]
name = parts[1]
unique_name = parts[2]
name_class = parts[3]

# Create a dictionary with the parsed data
if name_class == "scientific name" and taxid not in taxid_to_scientific_name:
if unique_name:
taxid_to_scientific_name[taxid] = unique_name
scientific_name_to_taxid[unique_name] = taxid
else:
taxid_to_scientific_name[taxid] = name
scientific_name_to_taxid[name] = taxid
elif name_class == "genbank common name" and taxid not in taxid_to_common_name:
taxid_to_common_name[taxid] = name
return taxid_to_scientific_name, taxid_to_common_name, scientific_name_to_taxid

build_full_cache(url=url, cache_dir=output_dir)

def process_and_save_taxdmp_in_memory(url, output_dir):
"""
Downloads and processes the taxdump file from the given URL,
and saves the parsed data into separate JSON files in the specified output directory.

Args:
url (str): The URL of the taxdump file to download and process.
output_dir (str): The directory where the parsed data will be saved.

Returns:
None
"""
# Ensure the output directory exists
if not os.path.exists(output_dir):
os.makedirs(output_dir)

# Step 1: Download and unzip the folder in memory
names_dmp_content = download_and_extract_zip_in_memory(url)

# Step 2: Parse the names.dmp content
taxid_to_scientific_name, taxid_to_common_name, scientific_name_to_taxid = (
parse_dmp_content(names_dmp_content)
)

# Step 3: Save the dictionaries to files
with open(
os.path.join(output_dir, "taxid_to_common_name.json"), "w", encoding="utf-8"
) as f:
json.dump(taxid_to_common_name, f, indent=4)

with open(
os.path.join(output_dir, "taxid_to_scientific_name.json"), "w", encoding="utf-8"
) as f:
json.dump(taxid_to_scientific_name, f, indent=4)

with open(
os.path.join(output_dir, "scientific_name_to_taxid.json"), "w", encoding="utf-8"
) as f:
json.dump(scientific_name_to_taxid, f, indent=4)



def load_json(file_path):
"""
Load JSON data from a file.

Args:
file_path (str): The path to the JSON file.

Returns:
dict: The loaded JSON data.

"""
with open(file_path, "r", encoding="utf-8") as f:
return json.load(f)

@click.command()
@click.option("--reload", '-r', is_flag=True, help="Reload NCBI taxonomy data")

def download_ncbi_taxonomy(reload=False):

"""
Load JSON data from a file.

Args:
file_path (str): The path to the JSON file.

Returns:
dict: The loaded JSON data as a dictionary.
"""
if reload or not os.path.exists(SCIENTIFIC_NAME_TO_TAXONID_PATH) or not os.path.exists(TAXON_SCIENTIFIC_NAME_PATH) or not os.path.exists(TAXON_COMMON_NAME_PATH):
process_and_save_taxdmp_in_memory(NCBI_TAXON_URL, OUTPUT_DIR_PATH)
@click.option(
"--reload", "-r", is_flag=True, help="Re-download even if already cached."
)
@click.option(
"--data-dir",
"-d",
type=click.Path(file_okay=False),
default=None,
help="Directory to store the taxonomy cache in. Defaults to the bkbit cache directory (override with BKBIT_DATA_DIR).",
)
def download_ncbi_taxonomy(reload=False, data_dir=None):
"""Pre-download the full NCBI taxonomy used by gff2jsonld (optional)."""
target = ensure_full_cache(reload=reload, cache_dir=data_dir)
if reload:
click.echo(f"Re-downloaded NCBI taxonomy to {target}")
else:
print("PRINT already downloaded")
click.echo(f"NCBI taxonomy cache ready at {target}")


if __name__ == "__main__":
download_ncbi_taxonomy()
download_ncbi_taxonomy()
Loading
Loading