Identifier mapping

NeKo lazily loads a cached reviewed-human UniProt table for gene-symbol and accession translation. Unrecognized identifiers can use a bounded live UniProt fallback; successful and failed lookups are memoized for the process.

from neko.inputs.identifier_mapping import to_genesymbol, to_uniprot

accession = to_uniprot("EGFR")
symbol = to_genesymbol("P00533")

Set NEKO_CACHE_DIR to choose the cache root. refresh_cache() explicitly refreshes the reviewed-human table.

looks_like_uniprot_accession

looks_like_uniprot_accession(value: str) -> bool

Return True if value matches the UniProt accession pattern.

Source code in neko/inputs/identifier_mapping.py

def looks_like_uniprot_accession(value: str) -> bool:
    """Return True if `value` matches the UniProt accession pattern."""

    return bool(value) and bool(_RE_UNIPROT_ACCESSION.match(value))

to_uniprot

to_uniprot(label: str, organism: str = HUMAN_TAXON_ID) -> str | None

Translate a gene symbol to its primary UniProt accession.

If label is not recognized as a gene symbol but already looks like a valid UniProt accession, it is returned unchanged. Returns None if no translation could be found.

Source code in neko/inputs/identifier_mapping.py

def to_uniprot(label: str, organism: str = HUMAN_TAXON_ID) -> str | None:
    """
    Translate a gene symbol to its primary UniProt accession.

    If `label` is not recognized as a gene symbol but already looks like a
    valid UniProt accession, it is returned unchanged. Returns None if no
    translation could be found.
    """

    if not label:
        return None

    _ensure_loaded()

    accession = _state['symbol_to_uniprot'].get(label)

    if accession:
        return accession

    # Already looks like a UniProt accession (not a gene symbol): no need to
    # ask the live API to translate it as if it were one, just echo it back.
    if looks_like_uniprot_accession(label):
        return label

    if label in _state['symbol_fallback_cache']:
        return _state['symbol_fallback_cache'][label]

    dest = 'UniProtKB-Swiss-Prot' if organism == HUMAN_TAXON_ID else 'UniProtKB'
    fallback = _fallback_translate({label}, 'Gene_Name', dest, organism)
    result = fallback.get(label)
    _state['symbol_fallback_cache'][label] = result

    return result

to_genesymbol

to_genesymbol(uniprot_id: str) -> str | None

Translate a UniProt accession to its primary gene symbol.

If uniprot_id is not recognized as an accession but does not look like one either (e.g. it is already a gene symbol), it is returned unchanged. Returns None if no translation could be found.

Source code in neko/inputs/identifier_mapping.py

def to_genesymbol(uniprot_id: str) -> str | None:
    """
    Translate a UniProt accession to its primary gene symbol.

    If `uniprot_id` is not recognized as an accession but does not look like
    one either (e.g. it is already a gene symbol), it is returned unchanged.
    Returns None if no translation could be found.
    """

    if not uniprot_id:
        return None

    _ensure_loaded()

    symbol = _state['uniprot_to_symbol'].get(uniprot_id)

    if symbol:
        return symbol

    # Does not look like a UniProt accession (presumably already a gene
    # symbol): no need to ask the live API to translate it as if it were
    # one, just echo it back.
    if not looks_like_uniprot_accession(uniprot_id):
        return uniprot_id

    if uniprot_id in _state['uniprot_fallback_cache']:
        return _state['uniprot_fallback_cache'][uniprot_id]

    fallback = _fallback_translate(
        {uniprot_id}, 'UniProtKB_AC-ID', 'Gene_Name', None,
    )
    result = fallback.get(uniprot_id)
    _state['uniprot_fallback_cache'][uniprot_id] = result

    return result

refresh_cache

refresh_cache(force: bool = True) -> None

Force a fresh download of the offline identifier mapping table.

Source code in neko/inputs/identifier_mapping.py

def refresh_cache(force: bool = True) -> None:
    """Force a fresh download of the offline identifier mapping table."""

    _ensure_loaded(force_refresh = force)

Batched translator

IDTranslator is the separate batched translation interface used by legacy workflows.

IDTranslator

IDTranslator(input_file: str, output_file: str, source_type: str, dest_type: str, pickle_dir: str = './pickles', batch_size: int = 100, processes: int = None, columns: List[str] = None, max_retries: int = 3, clear_progress: bool = False, input_columns: Union[List[str], Dict[str, str]] = None, has_header: bool = True, multiple_mapping_strategy: str = 'expand')

Source code in neko/inputs/db_translator.py

def __init__(self, input_file: str, output_file: str, source_type: str, dest_type: str,
             pickle_dir: str = './pickles', batch_size: int = 100, processes: int = None,
             columns: List[str] = None, max_retries: int = 3, clear_progress: bool = False,
             input_columns: Union[List[str], Dict[str, str]] = None, has_header: bool = True,
             multiple_mapping_strategy: str = 'expand'):
    self.input_file = input_file
    self.output_file = output_file
    self.source_type = source_type
    self.dest_type = dest_type
    self.pickle_dir = pickle_dir
    self.batch_size = batch_size
    self.processes = processes or max(1, cpu_count() - 1)
    self.columns = columns or ['source', 'target']
    self.max_retries = max_retries
    self.input_columns = input_columns
    self.has_header = has_header
    self.df = None
    self.id_mapping = {}
    self.unique_ids = set()
    self.multiple_mapping_strategy = multiple_mapping_strategy

    # Set up logging
    self.logger = logging.getLogger(f"{self.__class__.__name__}_{id(self)}")
    self.logger.setLevel(logging.INFO)

    # Create formatter
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

    # Create file handler
    fh = logging.FileHandler('id_translator.log')
    fh.setLevel(logging.INFO)
    fh.setFormatter(formatter)
    self.logger.addHandler(fh)

    # Create stream handler for notebook display
    class NotebookHandler(logging.Handler):
        def emit(self, record):
            msg = self.format(record)
            display(HTML(f"<pre>{msg}</pre>"))

    nh = NotebookHandler()
    nh.setLevel(logging.INFO)
    nh.setFormatter(formatter)
    self.logger.addHandler(nh)

    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

    os.makedirs(self.pickle_dir, exist_ok=True)

    if clear_progress:
        self.clear_all_progress()

Methods:

load_translated_dataframe

load_translated_dataframe(file_path: str)

Loads a translated DataFrame from a file.

Args: file_path (str): The path to the file containing the translated DataFrame.

Returns: pd.DataFrame: The loaded DataFrame.

Source code in neko/inputs/db_translator.py

def load_translated_dataframe(self, file_path: str):
    """
    Loads a translated DataFrame from a file.

    Args:
        file_path (str): The path to the file containing the translated DataFrame.

    Returns:
        pd.DataFrame: The loaded DataFrame.
    """
    file_extension = os.path.splitext(file_path)[1].lower()
    if file_extension == '.csv':
        self.df = pd.read_csv(file_path)
    elif file_extension in ['.xls', '.xlsx']:
        self.df = pd.read_excel(file_path)
    elif file_extension == '.tsv':
        self.df = pd.read_csv(file_path, sep='\t')
    else:
        raise ValueError(f"Unsupported file format: {file_extension}")
    self.logger.info(f"Loaded translated DataFrame from {file_path}")

translate_single_identifier

translate_single_identifier(identifier: str, source_type: str, dest_type: str, taxon_id: int = None, sleep_time: int = 3, replace_in_db: bool = False, file_path: str = None)

Translates a single identifier and updates the DataFrame if the translation is successful.

Args: identifier (str): The identifier to translate. source_type (str): The source type of the identifier. dest_type (str): The destination type for the translation. sleep_time (int): The time to wait (in seconds) for the request to complete. replace_in_db (bool): Flag to decide if the translation should replace the original in the database. file_path (str): Optional path to a file containing a pre-translated DataFrame to update.

Returns: List[str]: The translated identifiers.

Source code in neko/inputs/db_translator.py

def translate_single_identifier(self, identifier: str, source_type: str, dest_type: str, taxon_id: int = None, sleep_time: int = 3,
                                replace_in_db: bool = False, file_path: str = None):
    """
    Translates a single identifier and updates the DataFrame if the translation is successful.

    Args:
        identifier (str): The identifier to translate.
        source_type (str): The source type of the identifier.
        dest_type (str): The destination type for the translation.
        sleep_time (int): The time to wait (in seconds) for the request to complete.
        replace_in_db (bool): Flag to decide if the translation should replace the original in the database.
        file_path (str): Optional path to a file containing a pre-translated DataFrame to update.

    Returns:
        List[str]: The translated identifiers.
    """
    if file_path:
        self.load_translated_dataframe(file_path)

    translated_ids = []
    for attempt in range(self.max_retries):
        try:
            # Submit the request to the IdMappingClient
            if source_type == 'Gene_Name' and taxon_id:
                request = IdMappingClient.submit(source=source_type, dest=dest_type, ids={identifier},
                                                 taxon_id=taxon_id)
            else:
                request = IdMappingClient.submit(source=source_type, dest=dest_type, ids={identifier})

            # Wait for the request to complete
            time.sleep(sleep_time)

            # Process the result
            result = list(request.each_result())
            translated_ids = [m['to'] for m in result] if result else []
            break
        except Exception as e:
            if attempt == self.max_retries - 1:
                self.logger.error(f"Error processing {identifier} after {self.max_retries} attempts: {str(e)}")

    # Automatically select the first translation
    if translated_ids:
        translated_ids = [translated_ids[0]]

    # Print the translated entry
    print(f"Original Identifier: {identifier}")
    print(f"Translated Identifier(s): {translated_ids if translated_ids else 'No translation found'}")

    # Optionally replace the identifier in the DataFrame
    if replace_in_db and translated_ids:
        source_column = f"{self.columns[0]}_{dest_type}"
        target_column = f"{self.columns[1]}_{dest_type}"

        for column, translated_column in zip(self.columns, [source_column, target_column]):
            self.df[translated_column] = self.df.apply(
                lambda row: ';'.join(translated_ids) if row[column] == identifier else row[translated_column],
                axis=1
            )

    return translated_ids

save_translated_dataframe

save_translated_dataframe(file_path: str)

Saves the translated DataFrame to a file.

Args: file_path (str): The path where the translated DataFrame should be saved.

Source code in neko/inputs/db_translator.py

def save_translated_dataframe(self, file_path: str):
    """
    Saves the translated DataFrame to a file.

    Args:
        file_path (str): The path where the translated DataFrame should be saved.
    """
    file_extension = os.path.splitext(file_path)[1].lower()
    if file_extension == '.csv':
        self.df.to_csv(file_path, index=False)
    elif file_extension in ['.xls', '.xlsx']:
        self.df.to_excel(file_path, index=False)
    elif file_extension == '.tsv':
        self.df.to_csv(file_path, sep='\t', index=False)
    else:
        raise ValueError(f"Unsupported file format: {file_extension}")
    self.logger.info(f"Translated DataFrame saved to {file_path}")