Module query_pipeline

IndexPipeline

IndexPipeline(
    worklow: Path,
    workflow_profile: Path,
    config: dict,
    working_dir: Path,
)

Bases: Pipeline

The IndexPipeline class implements the indexing of the TronMake k-mer pipeline.

determine_final_index

determine_final_index()

Find index output of user-specified k-mer indexing method

run_pipeline

run_pipeline(slurm: bool = True) -> IndexPipelineResult

Execute indexing pipeline

IndexPipelineConfig

IndexPipelineConfig(
    samples: Path,
    method: str,
    kmer_size: float = 21,
    cutoff: int = 2,
    fpr: float = 0.05,
    verbose=True,
)

Bases: KmerPipelineConfig

Generate config for indexing modus

Parameters:

• samples

  (str) –

  Path to TSV files with samples to include in k-mer index.

• method

  (str) –

  k-mer method used for indexing.

• kmer_size

  (float, default: 21 ) –

  k-mer size of index. Defaults to 21.

• cutoff

  (int, default: 2 ) –

  A value to define solid and weak k-mers. k-mers with lower occurence are omitted from index. Defaults to 2.

• fpr

  (float, default: 0.05 ) –

  The theoretical false positive rate of bloom filters. Defaults to 0.05.

• verbose

  (bool, default: True ) –

  Print configuation details. Defaults to True.

IndexPipelineResult dataclass

IndexPipelineResult(index_path: str)

Data class to hold results from indexing pipeline

KmerPipelineConfig

KmerPipelineConfig(
    query: bool, indexing: bool, verbose=True
)

Bases: PipelineConfig

Class to represent a base TronMake k-mer pipeline configuration.

Parameters:

• query

  (bool) –

  Pipeline should run in query mode.

• indexing

  (bool) –

  Pipeline should run in indexing mode.

• verbose

  (bool, default: True ) –

  Print configuation details. Defaults to True.

Pipeline

Pipeline(
    worklow: Path,
    workflow_profile: Path,
    config: QueryPipelineConfig,
    working_dir: Path,
    target_rule: str = "all",
)

Generic representation of a snakemake pipeline. The pipeline class holds the path to workflow (snakefile), a configuration object, the working directory and the requested target rule. Upon exeuction the config object is written into a yaml file and the pipeline executed. The class pipeline is intended to be a reusable interface for specific pipelines.

Parameters:

• worklow

  (pathlib) –

  The path to the tronmake-kmer-pipeline Snakefile.

• config

  (QueryPipelineConfig) –

  Pipeline configuration object.

• working_dir

  (Path) –

  Working directory of pipeline.

• target_rule

  (str, default: 'all' ) –

  Target rule to request from workflow. Defaults to "all".

run

run(
    dryrun: bool = False, slurm: bool = True, cores: int = 8
) -> int

Build and execute pipeline

The run method implements the execution of the pipeline. It takes care of writing the config object into yaml file and constructing the snakemake shell command for execution in a subprocess.

Parameters:

• dryrun

  (bool, default: False ) –

  Execute pipeline without performing any operation. Defaults to False.

• slurm

  (bool, default: True ) –

  Execute pipeline with slurm support. Defaults to True.

• cores

  (int, default: 8 ) –

  Number of cores. Defaults to 8.

Returns:

• int ( int ) –

  description

PipelineConfig

PipelineConfig(verbose=True)

Class to represent a generic pipeline configuration.

Parameters:

• verbose

  (bool, default: True ) –

  Print configuation details. Defaults to True.

log_configuration

log_configuration()

Log configuration dictionary on command line

QueryPipeline

QueryPipeline(
    workflow: Path,
    workflow_profile: Path,
    config: QueryPipelineConfig,
    working_dir: Path,
    target_rule: str,
)

Bases: Pipeline

The QueryPipeline class implements the search mode of the TronMake k-mer pipeline.

determine_final_query

determine_final_query()

Based on selected methods in index manifest, find query output that could be created by pipeline

run_pipeline

run_pipeline(
    slurm: bool = True, cores: int = 8
) -> QueryPipelineResult

Execute query pipeline

QueryPipelineConfig

QueryPipelineConfig(
    index: Path,
    kmer_ratio: float,
    index_to_method_mapping: dict,
    verbose=True,
)

Bases: KmerPipelineConfig

Generate config for query modus

Parameters:

• index

  (Path) –

  Path to yaml based k-mer manifest file.

• kmer_ratio

  (float) –

  K-mer ratio used by raptor to determine presence/absence in indexed samples.

• methods

  (set) –

  k-mer methods of indices described in manifest.

• verbose

  (bool, default: True ) –

  Print configuation details. Defaults to True.

QueryPipelineResult dataclass

QueryPipelineResult(
    query_path: list[tuple[str, str, Path]],
)

Data class to hold results from query pipeline

Module database

k4neo metadata database

This module aims to replace the TinyDB based metadata database implementation used in k4neo annotation steps. The reason for this module is to enable parallelization of the annotation class. As TinyDB does not support multiple processes accessing the database, we decided to port this to SQLITE3. The new database module will not use ORM and will only be based on standard sqlite3 library. We plan to provide a thin wrapper on top that somehow acts as the old implementation and provides API compatibility for existing methods.

Todo

Better error and value handling

CreateDataBase

CreateDataBase(
    db_file: Path,
    data_set_file: Path,
    tissue_map: Path,
    test: bool = False,
    timeout: float = 30.0,
)

Bases: DataBase

Class to construct k4neo metadata database from k4neo index data.

Parameters:

• db_file

  (Path) –

  A path to a database file. Can be empty string, when test equals to true.

• data_set_file

  (Path) –

  A file listing documents to insert into database.

• tissue_map

  (Path) –

  A file with accepted tissue identifiers.

• test

  (bool, default: False ) –

  Establish database in-memory for testing. Defaults to False.

• timeout

  (float, default: 30.0 ) –

  Seconds before table is locked error is raised. Defaults to 30.0.

Source code in k4neo/database_sqlite/database.py

def __init__(
    self,
    db_file: pathlib.Path,
    data_set_file: pathlib.Path,
    tissue_map: pathlib.Path,
    test: bool = False,
    timeout: float = 30.0,
):
    """Generate metadata database

    Args:
        db_file (pathlib.Path): A path to a database file. Can be empty string, when test equals to true.
        data_set_file (pathlib.Path): A file listing documents to insert into database.
        tissue_map (pathlib.Path): A file with accepted tissue identifiers.
        test (bool, optional): Establish database in-memory for testing. Defaults to False.
        timeout (float, optional): Seconds before table is locked error is raised. Defaults to 30.0.
    """
    super().__init__(db_file, test=test, timeout=timeout)
    self.data_set_file = data_set_file
    self.tissue_map = tissue_map

create_static_tables

create_static_tables()

Create metadata database tables. Currently the database consists of 4 static tables

• sample_study_mapping: A table mapping samples to study ids
• tissue_map: A table mapping public tissue identifiers to k4neo tissue identifiers.
• samples: A table describing indexed samples.
• tissue_counts: A table with tissue counts per study, diasease and developmental_stage
• index_information:

Source code in k4neo/database_sqlite/database.py

def create_static_tables(self):
    """
    Create metadata database tables. Currently the database consists of 4 static tables

    * sample_study_mapping: A table mapping samples to study ids
    * tissue_map: A table mapping public tissue identifiers to k4neo tissue identifiers.
    * samples: A table describing indexed samples.
    * tissue_counts: A table with tissue counts per study, diasease and developmental_stage
    * index_information:

    """
    cursor = self.connection.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS sample_study_mapping (
            sample_name TEXT,
            study_id TEXT,
            PRIMARY KEY (sample_name, study_id)
        );
        """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS tissue_map (
            tissue_public TEXT PRIMARY KEY,
            tissue TEXT NOT NULL,
            subtissue TEXT
        );
        """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS samples (
            sample_name TEXT,
            study_id TEXT, 
            runs TEXT NOT NULL,
            tissue TEXT NOT NULL,
            developmental_stage TEXT NOT NULL,
            disease TEXT NOT NULL,
            sex TEXT,
            PRIMARY KEY (sample_name, study_id),
            FOREIGN KEY (tissue)
                REFERENCES tissue_map(tissue_public)
                ON UPDATE CASCADE
                ON DELETE RESTRICT
        );
        """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS tissue_counts (
            tissue TEXT NOT NULL,
            study_id TEXT NOT NULL,
            disease TEXT NOT NULL,
            developmental_stage TEXT NOT NULL,
            count INTEGER NOT NULL,
            PRIMARY KEY (tissue, study_id, disease, developmental_stage)
        );
        """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS aggregated_tissue_counts (
            tissue TEXT NOT NULL,
            disease TEXT NOT NULL,
            developmental_stage TEXT NOT NULL,
            count INTEGER NOT NULL,
            PRIMARY KEY (tissue, disease, developmental_stage)
        );
        """)

    self.connection.commit()
    cursor.close()

insert_sample_table

insert_sample_table(study_records: dict)

Insert records into sample table

Method to insert sample records batchwise into database

Returns:

• –

  None

Source code in k4neo/database_sqlite/database.py

def insert_sample_table(self, study_records: dict):
    """Insert records into sample table

    Method to insert sample records batchwise into database

    Returns:
        None

    """
    cursor = self.connection.cursor()
    cursor.executemany(
        """
        INSERT INTO samples (sample_name, study_id, runs, tissue, developmental_stage, disease, sex) 
        VALUES (:sample_name, :study_id, :runs, :tissue, :developmental_stage, :disease, :sex)
        """,
        study_records,
    )
    self.connection.commit()
    cursor.close()

insert_tissue_table

insert_tissue_table(tissue_records: dict)

Insert records into tissue table

Method to insert tissue records batchwise into database

Returns:

• –

  None

Source code in k4neo/database_sqlite/database.py

def insert_tissue_table(self, tissue_records: dict):
    """Insert records into tissue table

    Method to insert tissue records batchwise into database

    Returns:
        None

    """
    cursor = self.connection.cursor()
    cursor.executemany(
        """
        INSERT INTO tissue_map (tissue_public, tissue, subtissue) 
        VALUES (:tissue_public, :tissue, :subtissue)
        """,
        tissue_records,
    )
    self.connection.commit()
    cursor.close()

precomputations

precomputations()

Contains precomputations that would be an unnecessary overhead when computed always on the fly. Should be run after inserting all samples

Source code in k4neo/database_sqlite/database.py

def precomputations(self):
    """
    Contains precomputations that would be an unnecessary overhead when computed always on the fly. Should be run
    after inserting all samples
    """
    table = pd.read_sql(
        """SELECT s.study_id, s.disease, s.developmental_stage,
        t.tissue AS tissue
        FROM samples s
        JOIN tissue_map t
        ON s.tissue = t.tissue_public
        """,
        self.connection,
    )
    table = (
        table[["tissue", "developmental_stage", "disease", "study_id"]]
        .value_counts()
        .to_frame()
        .reset_index()
    )
    # Returns record in document format
    table.to_sql(name="tissue_counts", con=self.connection, if_exists="replace", index=False)
    logger.info(
        f"-> Added {len(table)} precomputed study-specific tissue counts for into database"
    )

    table = pd.read_sql(
        """SELECT s.study_id, s.disease, s.developmental_stage,
        t.tissue AS tissue
        FROM samples s
        JOIN tissue_map t
        ON s.tissue = t.tissue_public
        """,
        self.connection,
    )
    table = (
        table[["tissue", "developmental_stage", "disease"]]
        .value_counts()
        .to_frame()
        .reset_index()
    )
    # Returns record in document format
    table.to_sql(
        name="aggregated_tissue_counts", con=self.connection, if_exists="replace", index=False
    )
    logger.info(f"-> Added {len(table)} precomputed aggregated tissue counts for into database")

setup_db

setup_db()

Initialize when establishing database handle

Source code in k4neo/database_sqlite/database.py

def setup_db(self):
    """
    Initialize when establishing database handle
    """
    self.create_static_tables()
    logger.info("-> Adding tissue mapping into database")
    available_tissues = Parser.parse_tissuemap_into_document(self.tissue_map)
    available_tissues = validate_tissue_record(available_tissues)
    self.insert_tissue_table(available_tissues)

    logger.info("-> Adding samples into database")
    for study_id, study_annot, sample_count in self._parse_study_table():
        # If study is not in database parse table into document format
        study_elements = Parser.parse_sample_into_document(study_annot)
        for this_element in study_elements:
            this_element["study_id"] = study_id
        # Update samples with tissue mapping and add subtissue section
        study_elements = validate_sample_record(study_elements)
        if len(study_elements) != sample_count:
            logger.debug(
                f"Dropped {sample_count - len(study_elements)} samples because of validation"
            )
            sample_count = sample_count - len(study_elements)

        self._add_samples(study_id, study_elements, sample_count)

DataBase

DataBase(
    db_file: Path, test: bool = False, timeout: float = 30.0
)

Generic class representing the k4neo metadata database

Parameters:

• db_file

  (Path) –

  A path to a database file.

• test

  (bool, default: False ) –

  Establish database in-memory for testing. Defaults to False.

• timeout

  (float, default: 30.0 ) –

  How many seconds should we wait before raising error that table is locked. Defaults to 30s.

Source code in k4neo/database_sqlite/database.py

def __init__(self, db_file: pathlib.Path, test: bool = False, timeout: float = 30.0):
    """Parameter initialization

    Args:
        db_file (pathlib.Path):  A path to a database file.
        test (bool, optional): Establish database in-memory for testing. Defaults to False.
        timeout(float, optional): How many seconds should we wait before raising error that table is locked. Defaults to 30s.
    """
    self.db_file = db_file
    self.connection = None
    self.timeout = timeout
    self.test = test

connect

connect()

Establish connection and apply PRAGMA statements

Source code in k4neo/database_sqlite/database.py

def connect(self):
    """Establish connection and apply PRAGMA statements"""
    if self.test:
        self.connection = sqlite3.connect(
            ":memory:", timeout=self.timeout, check_same_thread=False
        )
        logger.info("Established in-memory database")
    else:
        self.connection = sqlite3.connect(
            self.db_file, timeout=self.timeout, check_same_thread=False
        )
        logger.info(f"Established connection to: {self.db_file}")
    self.connection.execute("PRAGMA journal_mode=WAL;")
    self.connection.execute("PRAGMA foreign_keys=ON;")