Outputs API

This module manages the file structure and tracking of Snakemake workflow outputs.

seqnado.outputs.SeqnadoOutputFiles

Bases: BaseModel

Collection of output files generated by SeqNado.

all_files property

all_files: List[str]

Return all files in the output collection.

bam_files property

bam_files

Get BAM files for the samples, using IP samples for IP-based assays.

has_consensus_peaks property

has_consensus_peaks

Check if consensus peaks are present in the output files.

select_files

select_files(
    suffix: str,
    include: str | None = None,
    exclude: str | None = None,
    must_include_all_patterns: bool = False,
    use_regex: bool = False,
    case_sensitive: bool = False,
) -> List[str]

Filter files by suffix and optional substring.

Parameters:

Name	Type	Description	Default
suffix	str	The file suffix to filter by (e.g. ".txt" or "csv").	required
include	str	Substring or regex pattern that must be present in the file path.	None
exclude	str	Substring or regex pattern that must NOT be present in the file path.	None
must_include_all_patterns	bool	If True, all include patterns must match (AND). If False, any match suffices (OR).	False
use_regex	bool	If True, treat include/exclude as regex patterns.	False
case_sensitive	bool	If True, matching is case-sensitive.	False

Returns: List[str]: A list of file paths matching the criteria.

Source code in seqnado/outputs/core.py

def select_files(
    self,
    suffix: str,
    include: str | None = None,
    exclude: str | None = None,
    must_include_all_patterns: bool = False,
    use_regex: bool = False,
    case_sensitive: bool = False,
) -> List[str]:
    """Filter files by suffix and optional substring.

    Args:
        suffix (str): The file suffix to filter by (e.g. ".txt" or "csv").
        include (str, optional): Substring or regex pattern that must be present in the file path.
        exclude (str, optional): Substring or regex pattern that must NOT be present in the file path.
        must_include_all_patterns (bool): If True, all include patterns must match (AND). If False, any match suffices (OR).
        use_regex (bool): If True, treat include/exclude as regex patterns.
        case_sensitive (bool): If True, matching is case-sensitive.
    Returns:
        List[str]: A list of file paths matching the criteria.
    """
    fs = FileSelector(self.files)
    return fs.select(
        suffix=suffix,
        includes=include,
        excludes=exclude,
        case_sensitive=case_sensitive,
        use_regex=use_regex,
        includes_all=must_include_all_patterns,
    )

select_bigwig_subtype

select_bigwig_subtype(
    method: PileupMethod = PileupMethod.DEEPTOOLS,
    scale: DataScalingTechnique = DataScalingTechnique.UNSCALED,
    spikein_method: str | None = None,
    assay: Assay | None = None,
    is_merged: bool = False,
    ip_only: bool = False,
)

Select bigWig files of a specific subtype.

Parameters:

Name	Type	Description	Default
method	PileupMethod	The pileup method to filter by.	DEEPTOOLS
scale	DataScalingTechnique	The scale method to filter by.	UNSCALED
spikein_method	str	The spike-in method to filter by orlando or with_input. Defaults to None.	None
assay	Assay	The assay type to filter by. Defaults to None.	None
is_merged	bool	If True, select merged (consensus) bigWigs only.	False
ip_only	bool	If True, exclude control/input samples (requires ip_sample_names to be set).	False

Returns:

Type	Description
	List[str]: A list of bigWig files matching the specified subtype.

Source code in seqnado/outputs/core.py

def select_bigwig_subtype(
    self,
    method: PileupMethod = PileupMethod.DEEPTOOLS,
    scale: DataScalingTechnique = DataScalingTechnique.UNSCALED,
    spikein_method: str | None = None,
    assay: Assay | None = None,
    is_merged: bool = False,
    ip_only: bool = False,
):
    """Select bigWig files of a specific subtype.

    Args:
        method (PileupMethod): The pileup method to filter by.
        scale (DataScalingTechnique): The scale method to filter by.
        spikein_method (str, optional): The spike-in method to filter by orlando or with_input. Defaults to None. 
        assay (Assay, optional): The assay type to filter by. Defaults to None.
        is_merged (bool): If True, select merged (consensus) bigWigs only.
        ip_only (bool): If True, exclude control/input samples (requires ip_sample_names to be set).

    Returns:
        List[str]: A list of bigWig files matching the specified subtype.
    """
    includes = [method.value, scale.value]
    if is_merged:
        includes.append("merged")
    if spikein_method is not None:
        includes.append(spikein_method)
    if assay is not None:
        includes.append(assay.value.lower())

    results = self.select_files(
        ".bigWig",
        include=includes,
        exclude="/geo_submission/",
        must_include_all_patterns=True,
        case_sensitive=False,
    )
    # Separate individual from merged: both paths contain the scale string,
    # so filter explicitly on the presence/absence of "/merged/".
    if is_merged:
        results = [f for f in results if "/merged/" in f]
    else:
        results = [f for f in results if "/merged/" not in f]

    # Optionally filter to IP samples only (exclude control/input samples)
    if ip_only and self.ip_sample_names:
        ip_set = set(self.ip_sample_names)
        results = [f for f in results if Path(f).stem in ip_set]

    return results

select_heatmap_matrix

select_heatmap_matrix(
    scale: DataScalingTechnique,
    method: PileupMethod = PileupMethod.DEEPTOOLS,
    is_merged: bool = False,
    spikein_method: str | None = None,
) -> str

Return the matrix path (temp file, not in self.files).

Source code in seqnado/outputs/core.py

def select_heatmap_matrix(
    self,
    scale: DataScalingTechnique,
    method: PileupMethod = PileupMethod.DEEPTOOLS,
    is_merged: bool = False,
    spikein_method: str | None = None,
) -> str:
    """Return the matrix path (temp file, not in self.files)."""
    return f"{self._heatmap_dir(scale, method, is_merged, spikein_method)}/matrix.mat.gz"

select_track_plots

select_track_plots(
    scale: DataScalingTechnique,
    method: PileupMethod = PileupMethod.DEEPTOOLS,
    is_merged: bool = False,
    spikein_method: str | None = None,
) -> list[str]

Select genome browser plot files for a specific method/scale/merged combination.

When scale is SPIKEIN and spikein_method is provided, only plots for that specific spike-in method are returned.

Source code in seqnado/outputs/core.py

def select_track_plots(
    self,
    scale: DataScalingTechnique,
    method: PileupMethod = PileupMethod.DEEPTOOLS,
    is_merged: bool = False,
    spikein_method: str | None = None,
) -> list[str]:
    """Select genome browser plot files for a specific method/scale/merged combination.

    When scale is SPIKEIN and spikein_method is provided, only plots for
    that specific spike-in method are returned.
    """
    prefix = "merged/" if is_merged else ""
    if scale == DataScalingTechnique.SPIKEIN and spikein_method is not None:
        target = f"track_plots/{prefix}{method.value}/spikein/{spikein_method}/"
    else:
        target = f"track_plots/{prefix}{method.value}/{scale.value}/"
    return [f for f in self.files if target in f]

← Back to API Overview