src.utils.demultiplexor_adapter package

Submodules

src.utils.demultiplexor_adapter.bcl2fastq_adapter module

This module defines the BclToFastqAdapter class, which is a demultiplexer adapter for converting BCL files to FASTQ format using the Illumina bcl2fastq tool.

It inherits from LoggerMixin and IDemultiplexorAdapter, providing logging capabilities and adhering to the demultiplexer adapter interface.

The adapter handles configuration validation, command construction, and execution of the bcl2fastq command.

class src.utils.demultiplexor_adapter.bcl2fastq_adapter.BclToFastqAdapter(config: dict[str, str], cmd_caller: callable | None = <built-in function system>, logger: ~logging.Logger = None)

Bases: LoggerMixin, IDemultiplexorAdapter

Adapter class to convert BCL files to FASTQ format by Illumina bcl2fastq tool.

This class manages the construction and execution of a demultiplexing command based on provided configuration parameters. It validates the configuration, constructs command-line arguments accordingly, and executes the demultiplexing process via a specified command caller.

config

Configuration parameters used for demultiplexing.

Type:

dict[str, str]

cmd_caller

Function used to execute system commands.

Type:

callable

logger

Logger instance for logging messages (inherited from LoggerMixin).

Type:

logging.Logger

static _check_config(config: dict[str, str]) → tuple[bool, str]

Validates the provided configuration dictionary.

Parameters:

config (dict[str, str]) – The configuration dictionary to validate.

Returns: tuple[bool, str] contains

flag (bool): True if the configuration is valid, False otherwise. message (str): A message indicating the validation result or describing missing keys.

_add_param(arguments: list, arg_name: str, config_key=None, is_flag: bool = False)

Adds a command-line argument to the list based on configuration and parameters.

Parameters:

• arguments (list) – The list of command-line arguments to which new arguments will be appended.

• arg_name (str) – The argument name, e.g., ‘–min-log-level’.

• config_key (str, optional) – The key to look up in the configuration dictionary. Defaults to None, in which case the argument name without ‘–’ is used.

• is_flag (bool, optional) – If True, adds only the flag (without a value) if the corresponding config is True. Defaults to False.

Note

If ‘is_flag’ is True and the configuration value for the key is True, appends ‘arg_name’ to the arguments list. Otherwise, if a value exists in the configuration for the key, appends both ‘arg_name’ and the string representation of the value to the list.

demultiplex() → None

Constructs and executes the demultiplexing command based on the current configuration.

Note

Relies on the ‘_add_param’ method to append arguments based on configuration values. Assumes ‘self.config’ contains all necessary configuration entries. Uses ‘self.cmd_caller’ to execute the command with the constructed arguments.

extract_barcodes(r1_path: PathLike, r2_path: PathLike | None = None) → PathLike

Extract barcode subsequences from input sequence file(s).

Processes the provided FASTA, FASTQ, or compressed (gzipped) input file(s) to identify and extract barcode sequences or subsequences.

Supports single-end (R1 only) or paired-end (R1 and R2) data.

Parameters:

• r1_path (PathLike[AnyStr]) – Path to the input file containing forward read sequences.

• r2_path (Optional[PathLike[AnyStr]]) – Optional path to the input file containing reversed read sequences (paired-end data). Defaults to None.

Returns:

Path to the output file containing the extracted barcode subsequences.

Return type:

PathLike[AnyStr]

Raises:

Various exceptions depending on file processing errors (not explicitly documented here). –

_abc_impl = <_abc._abc_data object>

_is_protocol = False

src.utils.demultiplexor_adapter.demultiplexor_adapter_factory module

This module provides a factory for creating different types of demultiplexor adapters.

It handles the creation and selection logic, abstracting away the specific implementation details.

class src.utils.demultiplexor_adapter.demultiplexor_adapter_factory.DemultiplexorAdapterFactory(adapter_types: list[type[T]], logger: Logger = None)

Bases: LoggerMixin, Generic[T]

Factory class for creating demultiplexor adapters.

This class manages the creation process, enabling different demultiplexing implementations.

static create_adapter(adapter_type_name: str, config: dict[str, str], logger: ~logging.Logger = None, caller: ~src.core.base.CommandExecutor | callable = <built-in function system>) → IDemultiplexorAdapter

Creates a demultiplexor adapter based on the provided type name.

Parameters:

• adapter_type_name (str) – The name of the adapter type to create.

• config (dict[str, str]) – A dict of implementation-based agruments.

• logger (logging.Logger) – A logger instance to handle adapter output.

• caller (Union[CommandExecutor, callable]) – Callable entity to execute system commands.

Returns:

A demultiplexor adapter object of the specified type.

Raises:

ValueError – If no adapter of the requested type is found.

get_adapter(adapter_type_name: str, config: dict[str, str], logger: ~logging.Logger = None, caller: ~src.core.base.CommandExecutor | callable = <built-in function system>) → IDemultiplexorAdapter

Retrieve an instance of a demultiplexor adapter matching the specified type name.

Searches through the available adapter types managed by this factory and attempts to instantiate one that matches the provided adapter_type_name.

If successful, returns the adapter instance; otherwise, logs an error and returns None.

Parameters:

• adapter_type_name (str) – The name of the adapter type to instantiate.

• config (dict[str, str]) – Configuration parameters for the adapter.

• logger (logging.Logger, optional) – Logger for logging messages. Defaults to None.

• caller (Union[CommandExecutor, callable], optional) – Callable used by the adapter for execution purposes. Defaults to os.system.

Returns:

An instance of the adapter if found and created successfully. None:

If no matching adapter type is found or creation fails.

Return type:

IDemultiplexorAdapter

Raises:

• None explicitly; errors during adapter –

• instantiation are caught and logged. –

src.utils.demultiplexor_adapter.i_demultiplexor_adapter module

This module defines the IDemultiplexorAdapter protocol, which specifies the interface for demultiplexing adapters.

It’s intended to be used for various demultiplexing methods, ensuring a consistent API.

class src.utils.demultiplexor_adapter.i_demultiplexor_adapter.IDemultiplexorAdapter(*args, **kwargs)

Bases: Protocol

Protocol for demultiplexing adapters. This defines the methods that any demultiplexing adapter must implement.

demultiplex()

Performs the demultiplexing operation. This should handle all aspects of the demultiplexing process, including configuration, command execution, and error handling.

extract_barcodes(r1_path: PathLike, r2_path: PathLike | None = None) → PathLike

Extract barcode subsequences from input sequence file(s).

Processes the provided FASTA, FASTQ, or compressed (gzipped) input file(s) to identify and extract barcode sequences or subsequences.

Supports single-end (R1 only) or paired-end (R1 and R2) data.

Parameters:

• r1_path (PathLike[AnyStr]) – Path to the input file containing forward read sequences.

• r2_path (Optional[PathLike[AnyStr]]) – Optional path to the input file containing reversed read sequences (paired-end data). Defaults to None.

Returns:

Path to the output file containing the extracted barcode subsequences.

Return type:

PathLike[AnyStr]

Raises:

Various exceptions depending on file processing errors (not explicitly documented here). –

_abc_impl = <_abc._abc_data object>

_is_protocol = True

Module contents