bend.utils package

Submodules

bend.utils.data_downstream module

data_downstream.py

Data loading and processing utilities for training downsteam tasks on embeddings saved in webdataset .tar.gz format.

bend.utils.data_downstream.collate_fn_pad_to_longest(batch, padding_value=-100)

Collate function for dataloader that pads to the longest sequence in the batch. :param batch: List of samples to collate. :type batch: list :param padding_value: Value to pad with. The default is -100. :type padding_value: int, optional

Returns:

padded – Padded batch.

Return type:

Tuple[torch.Tensor]

bend.utils.data_downstream.get_data(data_dir: str, train_data: List[str] | None = None, valid_data: List[str] | None = None, test_data: List[str] | None = None, cross_validation: bool | int = False, batch_size: int = 8, num_workers: int = 32, padding_value=-100, shuffle: int | None = None, **kwargs)

Function to get data from tar files.

Parameters:

• data_dir (str) – Path to data directory containing the tar files.

• train_data (List[str], optional) – List of paths to train tar files. The default is None. In case of cross validation can be simply the path to the data directory.

• valid_data (List[str], optional) – List of paths to valid tar files. The default is None.

• test_data (List[str], optional) – List of paths to test tar files. The default is None.

• cross_validation (Union[bool, int], optional) – If int, use the given partition as test set, +1 as valid set and the rest as train set. First split is 1. The default is False.

• batch_size (int, optional) – Batch size. The default is 8.

• num_workers (int, optional) – Number of workers for data loading. The default is 0.

• padding_value (int, optional) – Value to pad with. The default is -100.

• shuffle (int, optional) – Whether to shuffle the data. The default is None.

Returns:

• train_dataloader (torch.utils.data.DataLoader) – Dataloader for training data.

• valid_dataloader (torch.utils.data.DataLoader) – Dataloader for validation data.

• test_dataloader (torch.utils.data.DataLoader) – Dataloader for test data.

bend.utils.data_downstream.pad_to_longest(sequences: List[Tensor], padding_value=-100, batch_first=True)

Pad a list of sequences to the longest sequence in the list. :param sequences: List of sequences to pad. :type sequences: list[torch.Tensor] :param padding_value: Value to pad with. The default is -100. :type padding_value: int, optional :param batch_first: Whether to return the batch dimension first. The default is True. :type batch_first: bool, optional

Returns:

sequences – Padded sequences.

Return type:

torch.Tensor

bend.utils.data_downstream.return_dataloader(data: str | list, batch_size: int = 8, num_workers: int = 0, padding_value=-100, shuffle: int | None = None)

Function to return a dataloader from a list of tar files or a single one.

Parameters:

• data (Union[str, list]) – Path to single tar file or list of paths to tar files.

• batch_size (int, optional) – Batch size. The default is 8.

• num_workers (int, optional) – Number of workers for data loading. The default is 0.

• padding_value (int, optional) – Value to pad with. The default is -100.

• shuffle (int, optional) – Whether to shuffle the data. The default is None.

bend.utils.data_downstream.worker_init_fn(self, _)

Initialize worker function for data loading to make sure that each worker loads a different part of the data. See the pytorch data loading documentation for more information.

bend.utils.download module

download.py

Utility functions for downloading pretrained models from the web.

bend.utils.download.download_model(model: str = 'convnet', base_url: str = 'https://sid.erda.dk/share_redirect/dbQM0pgSlM/pretrained_models/', destination_dir: str = './pretrained_models/') → None

Download BEND pretrained model checkpoints from the ERDA URL. Uses wget to download the files.

Parameters:

• model (str) – Model to download. Needs to be a directory name in base_url.

• base_url (str) – Base URL to download from. Default is BEND’s pretrained models directory on ERDA.

• destination_dir (str) – Destination directory to download to. Default is ./pretrained_models/

Return type:

None.

bend.utils.download.download_model_zenodo(base_url: str, destination_dir: str = './pretrained_models')

Download a HF model hosted as a Zenodo record. Uses wget to download the files. We use this to get the GROVER model, but it should work for any model hosted on Zenodo as a flat directory.

Parameters:

• base_url (str) – Base URL to download from.

• destination_dir (str) – Destination directory to download to. Default is ./pretrained_models/

Return type:

None.

bend.utils.retrieve_from_bed module

retrieve_from_bed.py

Class to extract sequences from a reference genome using a bed file of genomic coordinates.

Example

get_dna = Annotation(annotation = 'path/to/bed/file', reference_genome = '/path/to/genome/fasta')

get_dna.get_dna_segment(index = 0) # will return the dna segment for index 0 in the annotation file

class bend.utils.retrieve_from_bed.Annotation(annotation: str | None = None, reference_genome: str | None = None)

Bases: object

An annotation object that can be used to retrieve DNA segments from a reference genome.

Get an Annotation object that can retrieve sequences from a reference genome.

Parameters:

• annotation (str, optional) – Path to a bed file containing genomic coordinates. The default is None.

• reference_genome (str, optional) – Path to a reference genome fasta file. The default is None.

extend_segments(extra_context_left: int | None = None, extra_context_right: int | None = None, extra_context: int | None = None) → None

Add extra context to the coordinates in the annotation file. Each sample in the annotation file will be extended by extra_context_left and extra_context_right.

Parameters:

• extra_context_left (int, optional) – Number of nucleotides to add to the left of each segment. The default is None.

• extra_context_right (int, optional) – Number of nucleotides to add to the right of each segment. The default is None.

• extra_context (int, optional) – Number of nucleotides to add to both sides of each segment. Use this instead of extra_context_left and extra_context_right. The default is None.

Raises:

ValueError – If extra_context is used simultaneously with extra_context_left or extra_context_right.

Return type:

None.

get_dna_segment(index) → str

Get a DNA sequence from the reference genome for a segment.

Parameters:

index (int) – Index of the row in the bed file for which to return the DNA sequence.

Returns:

dna_segment – The geomic DNA sequence of the segment.

Return type:

str

get_item(index: int)

Get a row from the annotation file.

Parameters:

index (int) – Index of the row to return.

Returns:

row – Row of the annotation file.

Return type:

pandas.Row

bend.utils.sequences module

class bend.utils.sequences.EncodeSequence(nucleotide_categories=['A', 'C', 'G', 'N', 'T'])

Bases: object

Encode or decode sequence into integers, onehot or string. :param nucleotide_categories: List with nucleotide categories, by default categories_4_letters_unknown :type nucleotide_categories: list

inverse_transform_integer(sequence)

Decode integer encoded sequence into string. :param sequence: Encoded sequence. :type sequence: np.ndarray

Returns:

sequence – Decoded sequence.

Return type:

str

static reduce_last_dim(sequence)

Reduce last dimension of sequence. :param sequence: Sequence to reduce last dimension. :type sequence: np.ndarray

Returns:

sequence – Sequence with reduced last dimension.

Return type:

np.ndarray

transform_integer(sequence, return_onehot=False)

Encode string nucleotide sequence into integers or onehot. :param sequence: Sequence to encode. :type sequence: str, list, np.ndarray :param return_onehot: Return onehot encoded sequence, by default False :type return_onehot: bool, optional

Returns:

sequence – Encoded sequence.

Return type:

np.ndarray

bend.utils.sequences.count_nucleotides(fasta, destination=None)

Count occurence of each nucleotide in fasta file. :param fasta: Path to fasta file. :type fasta: str :param destination: Path to save dictionary with counts, by default None :type destination: str, optional

Returns:

counts – Dictionary with counts.

Return type:

dict

bend.utils.task_trainer module

task_trainer.py

Trainer class for training downstream models on supervised tasks.

class bend.utils.task_trainer.BCEWithLogitsLoss(class_weights: Tensor | None = None, reduction: str = 'none')

Bases: Module

BCEWithLogitsLoss for classification tasks. Wrapper around torch.nn.BCEWithLogitsLoss that takes care of the dimensionality of the input and target tensors.

Get a BCEWithLogitsLoss object that can be used to train a model. :param class_weights: Weight for positive class :type class_weights: torch.Tensor

forward(pred, target, padding_value=-100)

Calculate the BCEWithLogitsLoss for a given prediction and target.

Parameters:

• pred (torch.Tensor) – Prediction tensor of logits.

• target (torch.Tensor) – Target tensor of labels.

• padding_value (int, optional) – Value to ignore in the loss calculation. The default is -100.

Returns:

loss – BCEWithLogitsLoss.

Return type:

torch.Tensor

class bend.utils.task_trainer.BaseTrainer(model, optimizer, criterion, device, config, overwrite_dir=False, gradient_accumulation_steps: int = 1)

Bases: object

‘Performs training and validation steps for a given model and dataset. We use hydra to configure the trainer. The configuration is passed to the trainer as an OmegaConf object.

Get a BaseTrainer object that can be used to train a model.

Parameters:

• model (torch.nn.Module) – Model to train.

• optimizer (torch.optim.Optimizer) – Optimizer to use for training.

• criterion (torch.nn.Module) – Loss function to use for training.

• device (torch.device) – Device to use for training.

• config (OmegaConf) – Configuration object.

• overwrite_dir (bool, optional) – Whether to overwrite the output directory. The default is False.

• gradient_accumulation_steps (int, optional) – Number of gradient accumulation steps. The default is 1.

test(test_loader, checkpoint=None, overwrite=False)

Performs testing.

Parameters:

• test_loader (torch.utils.data.DataLoader) – The data loader to be used.

• checkpoint (pandas.DataFrame, optional) – The checkpoint to be used. If None, loads the checkpoint with the lowest validation loss.

• overwrite (bool, optional) – If True, overwrites the best_model_metrics file.

Returns:

• loss (float) – The average validation loss.

• metric (float) – The average validation metric.

train(train_loader, val_loader, test_loader, epochs, load_checkpoint: bool | int = True)

Performs the full training routine.

Parameters:

• train_loader (torch.utils.data.DataLoader) – The training data loader.

• val_loader (torch.utils.data.DataLoader) – The validation data loader. epochs : int The number of epochs to train for.

• load_checkpoint (bool, optional) – If True, loads the latest checkpoint from the output directory and continues training. If an integer is provided, loads the checkpoint from that epoch and continues training.

Return type:

None

train_epoch(train_loader)

Performs one epoch of training.

Parameters:

train_loader (torch.utils.data.DataLoader) – The training data loader.

Returns:

train_loss – The average training loss for the epoch.

Return type:

float

train_step(batch, idx=0)

Performs a single training step.

Parameters:

• batch (tuple) – A tuple containing the batch of data and labels, as returned by the data loader.

• idx (int) – The index of the batch.

Returns:

loss – The loss for the batch.

Return type:

float

validate(data_loader)

Performs validation.

Parameters:

data_loader (torch.utils.data.DataLoader) – The data loader to be used.

Returns:

• loss (float) – The average validation loss.

• metrics (list) – The values of the validation metrics.

class bend.utils.task_trainer.CrossEntropyLoss(ignore_index=-100, weight=None)

Bases: Module

Cross entropy loss for classification tasks. Wrapper around torch.nn.CrossEntropyLoss that takes care of the dimensionality of the input and target tensors.

Get a CrossEntropyLoss object that can be used to train a model.

Parameters:

• ignore_index (int, optional) – Index to ignore in the loss calculation. Passed to torch.nn.CrossEntropyLoss. The default is -100.

• weight (torch.Tensor, optional) – Weights to apply to the loss. Passed to torch.nn.CrossEntropyLoss. The default is None.

forward(pred, target)

Calculate the cross entropy loss for a given prediction and target.

Parameters:

• pred (torch.Tensor) – Prediction tensor of logits.

• target (torch.Tensor) – Target tensor of labels.

Returns:

loss – Cross entropy loss.

Return type:

torch.Tensor

class bend.utils.task_trainer.MSELoss

Bases: Module

MSE loss for regression tasks. Wrapper around torch.nn.MSELoss that takes care of the dimensionality of the input and target tensors.

Get a MSELoss object that can be used to train a model.

forward(pred, target)

Calculate the MSE loss for a given prediction and target.

Parameters:

• pred (torch.Tensor) – Prediction tensor.

• target (torch.Tensor) – Target tensor.

Returns:

loss – MSE loss.

Return type:

torch.Tensor

class bend.utils.task_trainer.PoissonLoss

Bases: Module

Poisson loss for regression tasks.

Get a PoissonLoss object that can be used to train a model.

forward(pred, target)

Calculate the poisson loss for a given prediction and target.

Parameters:

• pred (torch.Tensor) – Prediction tensor.

• target (torch.Tensor) – Target tensor.

Returns:

loss – Poisson loss.

Return type:

torch.Tensor

Module contents

bend.utils

This module contains a collection of utilities used throughout the project for data processing, model training, and evaluation.

• Annotation: a class for retrieving sequences from a reference genome based on a bed file.

• TaskTrainer: a class for training a model on a given task.