kg_microbe.transform_utils.bakta package

Submodules

kg_microbe.transform_utils.bakta.bakta module

Transform for Bakta genome annotations.

class kg_microbe.transform_utils.bakta.bakta.BaktaTransform(input_dir=None, output_dir=None)

Bases: Transform

Transform Bakta genome annotations into KGX format.

add_cog_annotation(gene_id, cog_id)

Add COG node and gene -> COG edge.

Parameters:

gene_id (str) – Gene identifier
cog_id (str) – COG identifier (e.g., ‘COG:COG0665’)

Return type:

None

add_ec_annotation(protein_id, ec_id)

Add EC number node and protein -> EC edge.

Parameters:

protein_id (str) – Protein identifier
ec_id (str) – EC identifier (e.g., ‘EC:1.4.99.6’)

Return type:

None

add_edge(subject, predicate, obj, relation)

Add an edge to the collection.

Parameters:

subject (str) – Subject node ID
predicate (str) – Biolink predicate
obj (str) – Object node ID
relation (str) – RO or other relation ontology term

Return type:

None

add_functional_annotations(protein_id, gene_id, annotations)

Add functional annotation nodes and edges.

Parameters:

protein_id (str) – Protein identifier
gene_id (str) – Gene identifier
annotations (Dict[str, List[str]]) – Parsed annotations dictionary

Return type:

None

add_gene_node(gene_id, gene_symbol, product, annotations)

Add a gene node.

Parameters:

gene_id (str) – Composite gene ID
gene_symbol (str) – Gene symbol (may be empty)
product (str) – Gene product description
annotations (Dict[str, List[str]]) – Parsed annotations dictionary

Return type:

None

add_go_annotation(protein_id, go_id)

Add GO term node and protein -> GO edge.

Parameters:

protein_id (str) – Protein identifier
go_id (str) – GO identifier (e.g., ‘GO:0003677’)

Return type:

None

add_kegg_annotation(gene_id, kegg_id)

Add KEGG KO node and gene -> KEGG edge.

Parameters:

gene_id (str) – Gene identifier
kegg_id (str) – KEGG identifier (e.g., ‘KEGG:K19746’)

Return type:

None

add_organism_node(organism_id, samn_id)

Add a strain-level organism node (SAMN) and link to species (NCBITaxon).

Parameters:

organism_id (str) – SAMN organism identifier (e.g., ‘SAMN:00139461’)
samn_id (str) – SAMN identifier for lookup

Return type:

None

add_protein_node(protein_id, product, annotations)

Add a protein node.

Parameters:

protein_id (str) – Protein identifier (RefSeq or UniRef)
product (str) – Protein product description
annotations (Dict[str, List[str]]) – Parsed annotations dictionary

Return type:

None

get_ncbitaxon_id(samn_id)

Get NCBITaxon ID for a SAMN identifier.

Parameters:: samn_id (str) – SAMN identifier
Return type:: Optional[str]
Returns:: NCBITaxon ID or None

get_organism_id(samn_id)

Get the organism ID (SAMN-level for strain resolution).

Parameters:: samn_id (str) – SAMN identifier
Return type:: str
Returns:: SAMN organism ID (e.g., ‘SAMN:00139461’)

process_gene(gene_data, samn_id, organism_id)

Process a single gene annotation.

Parameters:

gene_data (Dict[str, str]) – Gene data dictionary from Bakta TSV
samn_id (str) – SAMN identifier
organism_id (Optional[str]) – Organism identifier (if available)

Return type:

None

process_genome(samn_dir, samn_id)

Process a single genome directory.

Parameters:

samn_dir (Path) – Path to SAMN directory
samn_id (str) – SAMN identifier

Return type:

None

run(data_file=None, show_status=True)

Run the Bakta transform.

Processes all dataset subdirectories in data/raw/bakta/ (e.g., pfas_bakta, cmm_bakta) and outputs to data/transformed/bakta/[dataset_name]/ for each.

Parameters:

data_file (Optional[Path]) – Not used (kept for API compatibility)
show_status (bool) – Show progress bar (default: True)

Return type:

None

write_output(dataset_name)

Write nodes and edges to TSV files for a specific dataset.

Parameters:: dataset_name (str) – Name of the dataset (e.g., ‘pfas_bakta’, ‘cmm_bakta’)
Return type:: None

kg_microbe.transform_utils.bakta.create_samn_mapping module

Helper script to create SAMN to NCBITaxon mapping file.

This script can be used to generate the samn_to_ncbitaxon.tsv mapping file needed by the Bakta transform. It queries NCBI databases to map BioSample accessions (SAMN) to NCBI Taxonomy IDs.

Usage:: python create_samn_mapping.py –input bakta_dir –output samn_to_ncbitaxon.tsv
Requirements:: pip install biopython

Note: This requires network access to query NCBI Entrez API. You may need to set NCBI_API_KEY environment variable for higher rate limits.

kg_microbe.transform_utils.bakta.create_samn_mapping.create_mapping_file(samn_ids, output_file, email, delay=0.4)

Create SAMN to NCBITaxon mapping file by querying NCBI.

Parameters:

samn_ids (list) – List of SAMN IDs to map
output_file (Path) – Path to output TSV file
email (str) – Email for NCBI Entrez
delay (float) – Delay between queries in seconds (default: 0.4s for 3 requests/sec)

Return type:

Dict[str, str]

Returns:

Dictionary of successfully mapped IDs

kg_microbe.transform_utils.bakta.create_samn_mapping.get_all_samn_ids(bakta_dir)

Extract all SAMN IDs from Bakta directory structure.

Parameters:: bakta_dir (Path) – Path to bakta directory containing SAMN subdirectories
Return type:: list
Returns:: List of SAMN IDs

kg_microbe.transform_utils.bakta.create_samn_mapping.load_existing_mapping(mapping_file)

Load existing mapping file to avoid re-querying.

Parameters:: mapping_file (Path) – Path to existing mapping file
Return type:: Dict[str, str]
Returns:: Dictionary of SAMN to NCBITaxon mappings

kg_microbe.transform_utils.bakta.create_samn_mapping.main(): Run the SAMN to NCBITaxon mapping creation process.

kg_microbe.transform_utils.bakta.create_samn_mapping.query_ncbi_for_taxon(samn_id, email='your_email@example.com')

Query NCBI to get NCBITaxon ID for a BioSample accession.

Parameters:

samn_id (str) – BioSample accession (e.g., ‘SAMN00139461’)
email (str) – Email for NCBI Entrez (required by NCBI)

Return type:

Optional[str]

Returns:

NCBITaxon ID or None if not found

kg_microbe.transform_utils.bakta.utils module

Utility functions for Bakta genome annotations transform.

kg_microbe.transform_utils.bakta.utils.create_gene_id(samn_id, locus_tag)

Create a composite gene ID from SAMN ID and locus tag.

Parameters:

samn_id (str) – SAMN identifier (e.g., ‘SAMN00139461’)
locus_tag (str) – Gene locus tag (e.g., ‘JEECHJ_00005’)

Return type:

str

Returns:

Composite gene ID (e.g., ‘SAMN00139461:JEECHJ_00005’)

kg_microbe.transform_utils.bakta.utils.extract_samn_from_path(path)

Extract SAMN ID from directory or file path.

Parameters:: path (Path) – Path containing SAMN ID
Return type:: Optional[str]
Returns:: SAMN ID (e.g., ‘SAMN00139461’) or None if not found

kg_microbe.transform_utils.bakta.utils.find_bakta_tsv(samn_dir)

Find the .bakta.tsv file in a SAMN directory.

Parameters:: samn_dir (Path) – Path to SAMN directory
Return type:: Optional[Path]
Returns:: Path to .bakta.tsv file or None if not found

kg_microbe.transform_utils.bakta.utils.get_all_samn_directories(base_dir)

Get all SAMN directories in the base Bakta directory.

Parameters:: base_dir (Path) – Base directory containing SAMN subdirectories
Return type:: List[Path]
Returns:: List of paths to SAMN directories

kg_microbe.transform_utils.bakta.utils.get_biolink_category_for_go(aspect)

Map GO aspect to Biolink category.

Parameters:: aspect (str) – GO aspect (‘biological_process’, ‘molecular_function’, ‘cellular_component’)
Return type:: str
Returns:: Biolink category string

kg_microbe.transform_utils.bakta.utils.get_biolink_predicate_for_go(aspect)

Get Biolink predicate and RO relation for a GO aspect.

Parameters:: aspect (str) – GO aspect (‘biological_process’, ‘molecular_function’, ‘cellular_component’)
Return type:: Tuple[str, str]
Returns:: Tuple of (predicate, relation)

kg_microbe.transform_utils.bakta.utils.get_go_aspect(go_id, go_adapter=None, cache=None, *, raw_dir=None, authority=None)

Determine the aspect (namespace) of a GO term.

Reads via the raw-sqlite namespace map exposed by kg_microbe.utils.ontology_utils.get_go_aspect — OAK’s entity_metadata_map throws on GO databases with case-collision prefixes, and the swallowed exception used to leave aspect at the molecular_function default, silently miscategorising every biological-process and cellular-component term. An unusable GO DB raises OntologyDbUnavailableError (a BaseException), which is intentionally not catchable as Exception here.

Parameters:

go_id (str) – GO identifier (e.g., ‘GO:0003677’)
go_adapter – Unused; kept for backward compatibility with existing callers.
cache (Optional[Dict[str, str]]) – Optional cache dictionary for GO term → aspect mappings.
raw_dir – Explicit producer raw root; use the authority’s fingerprinted cache.
authority – Prepared immutable authority, avoiding ontology IO entirely.

Return type:

str

Returns:

One of ‘biological_process’, ‘molecular_function’, ‘cellular_component’.

kg_microbe.transform_utils.bakta.utils.get_protein_id(annotations, prefer_refseq=True)

Get the best protein identifier from annotations.

Strategy: Prefer RefSeq, fall back to UniRef50.

Parameters:

annotations (Dict[str, List[str]]) – Dictionary of parsed annotations from parse_dbxrefs()
prefer_refseq (bool) – Whether to prefer RefSeq over UniRef (default: True)

Return type:

Optional[str]

Returns:

Protein ID string with prefix, or None if no protein ID found

kg_microbe.transform_utils.bakta.utils.load_samn_to_ncbitaxon_mapping(mapping_file)

Load SAMN to NCBITaxon mapping from TSV file.

Parameters:: mapping_file (Path) – Path to TSV file with columns: samn_id, ncbitaxon_id
Return type:: Dict[str, str]
Returns:: Dictionary mapping SAMN IDs to NCBITaxon IDs

kg_microbe.transform_utils.bakta.utils.parse_bakta_tsv(tsv_file, feature_types=None)

Parse a Bakta TSV annotation file.

Parameters:

tsv_file (Path) – Path to .bakta.tsv file
feature_types (Optional[Set[str]]) – Set of feature types to include (default: {‘cds’})

Return type:

List[Dict[str, str]]

Returns:

List of dictionaries with gene annotations

kg_microbe.transform_utils.bakta.utils.parse_dbxrefs(dbxref_string)

Parse the DbXrefs column from Bakta TSV into structured annotations.

Parameters:: dbxref_string (str) – Comma-separated string of database cross-references
Return type:: Dict[str, List[str]]
Returns:: Dictionary with annotation types as keys and lists of IDs as values

Module contents

Bakta genome annotations transform.

class kg_microbe.transform_utils.bakta.BaktaTransform(input_dir=None, output_dir=None)

Bases: Transform

Transform Bakta genome annotations into KGX format.

add_cog_annotation(gene_id, cog_id)

Add COG node and gene -> COG edge.

Parameters:

gene_id (str) – Gene identifier
cog_id (str) – COG identifier (e.g., ‘COG:COG0665’)

Return type:

None

add_ec_annotation(protein_id, ec_id)

Add EC number node and protein -> EC edge.

Parameters:

protein_id (str) – Protein identifier
ec_id (str) – EC identifier (e.g., ‘EC:1.4.99.6’)

Return type:

None

add_edge(subject, predicate, obj, relation)

Add an edge to the collection.

Parameters:

subject (str) – Subject node ID
predicate (str) – Biolink predicate
obj (str) – Object node ID
relation (str) – RO or other relation ontology term

Return type:

None

add_functional_annotations(protein_id, gene_id, annotations)

Add functional annotation nodes and edges.

Parameters:

protein_id (str) – Protein identifier
gene_id (str) – Gene identifier
annotations (Dict[str, List[str]]) – Parsed annotations dictionary

Return type:

None

add_gene_node(gene_id, gene_symbol, product, annotations)

Add a gene node.

Parameters:

gene_id (str) – Composite gene ID
gene_symbol (str) – Gene symbol (may be empty)
product (str) – Gene product description
annotations (Dict[str, List[str]]) – Parsed annotations dictionary

Return type:

None

add_go_annotation(protein_id, go_id)

Add GO term node and protein -> GO edge.

Parameters:

protein_id (str) – Protein identifier
go_id (str) – GO identifier (e.g., ‘GO:0003677’)

Return type:

None

add_kegg_annotation(gene_id, kegg_id)

Add KEGG KO node and gene -> KEGG edge.

Parameters:

gene_id (str) – Gene identifier
kegg_id (str) – KEGG identifier (e.g., ‘KEGG:K19746’)

Return type:

None

add_organism_node(organism_id, samn_id)

Add a strain-level organism node (SAMN) and link to species (NCBITaxon).

Parameters:

organism_id (str) – SAMN organism identifier (e.g., ‘SAMN:00139461’)
samn_id (str) – SAMN identifier for lookup

Return type:

None

add_protein_node(protein_id, product, annotations)

Add a protein node.

Parameters:

protein_id (str) – Protein identifier (RefSeq or UniRef)
product (str) – Protein product description
annotations (Dict[str, List[str]]) – Parsed annotations dictionary

Return type:

None

get_ncbitaxon_id(samn_id)

Get NCBITaxon ID for a SAMN identifier.

Parameters:: samn_id (str) – SAMN identifier
Return type:: Optional[str]
Returns:: NCBITaxon ID or None

get_organism_id(samn_id)

Get the organism ID (SAMN-level for strain resolution).

Parameters:: samn_id (str) – SAMN identifier
Return type:: str
Returns:: SAMN organism ID (e.g., ‘SAMN:00139461’)

process_gene(gene_data, samn_id, organism_id)

Process a single gene annotation.

Parameters:

gene_data (Dict[str, str]) – Gene data dictionary from Bakta TSV
samn_id (str) – SAMN identifier
organism_id (Optional[str]) – Organism identifier (if available)

Return type:

None

process_genome(samn_dir, samn_id)

Process a single genome directory.

Parameters:

samn_dir (Path) – Path to SAMN directory
samn_id (str) – SAMN identifier

Return type:

None

run(data_file=None, show_status=True)

Run the Bakta transform.

Processes all dataset subdirectories in data/raw/bakta/ (e.g., pfas_bakta, cmm_bakta) and outputs to data/transformed/bakta/[dataset_name]/ for each.

Parameters:

data_file (Optional[Path]) – Not used (kept for API compatibility)
show_status (bool) – Show progress bar (default: True)

Return type:

None

write_output(dataset_name)

Write nodes and edges to TSV files for a specific dataset.

Parameters:: dataset_name (str) – Name of the dataset (e.g., ‘pfas_bakta’, ‘cmm_bakta’)
Return type:: None