Skip to content

Application

Classes

Modules

consensus_application

Provide an abstract base class for consensus building applications.

Classes

ConsensusApplication

Bases: ABC

Define an abstract base class for consensus building applications.

Source code in src/taxpasta/application/consensus_application.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class ConsensusApplication(ABC):
    """Define an abstract base class for consensus building applications."""

    @classmethod
    @abstractmethod
    def run(
        cls, profiles: Iterable[DataFrame[StandardProfile]], taxonomy: Path
    ) -> DataFrame:
        """
        Build a consensus from two or more taxonomic profiles.

        Args:
            profiles: Standardized profiles.
            taxonomy: Provide a shared taxonomy.

        """
Functions
run(profiles: Iterable[DataFrame[StandardProfile]], taxonomy: Path) -> DataFrame abstractmethod classmethod

Build a consensus from two or more taxonomic profiles.

Parameters:

Name Type Description Default
profiles Iterable[DataFrame[StandardProfile]]

Standardized profiles.

 required
taxonomy Path

Provide a shared taxonomy.

 required
Source code in src/taxpasta/application/consensus_application.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
@classmethod
@abstractmethod
def run(
    cls, profiles: Iterable[DataFrame[StandardProfile]], taxonomy: Path
) -> DataFrame:
    """
    Build a consensus from two or more taxonomic profiles.

    Args:
        profiles: Standardized profiles.
        taxonomy: Provide a shared taxonomy.

    """

error

Classes

Modules

standardisation_error

Provide an exception for errors during profile extraction and transformation.

Classes
StandardisationError

Bases: TaxpastaError

Define an exception for errors during profile extraction and transformation.

Source code in src/taxpasta/application/error/standardisation_error.py 
27
28
29
30
31
32
33
34
35
class StandardisationError(TaxpastaError):
    """Define an exception for errors during profile extraction and transformation."""

    def __init__(self, *, sample: str, profile: Path, message: str, **kwargs) -> None:
        """Initialize a taxpasta standardisation error."""
        super().__init__(**kwargs)
        self.sample = sample
        self.profile = profile
        self.message = message
Attributes
message = message instance-attribute
profile = profile instance-attribute
sample = sample instance-attribute
Functions
__init__(*, sample: str, profile: Path, message: str, **kwargs: str) -> None

Initialize a taxpasta standardisation error.

Source code in src/taxpasta/application/error/standardisation_error.py 
30
31
32
33
34
35
def __init__(self, *, sample: str, profile: Path, message: str, **kwargs) -> None:
    """Initialize a taxpasta standardisation error."""
    super().__init__(**kwargs)
    self.sample = sample
    self.profile = profile
    self.message = message
taxpasta_error

Provide the base class for all taxpasta application errors.

Classes
TaxpastaError

Bases: Exception

Define the base class for all taxpasta application errors.

Source code in src/taxpasta/application/error/taxpasta_error.py 
22
23
24
25
26
27
class TaxpastaError(Exception):
    """Define the base class for all taxpasta application errors."""

    def __int__(self, **kwargs) -> None:
        """Initialize a base taxpasta error."""
        super().__init__(**kwargs)
Functions
__int__(**kwargs) -> None

Initialize a base taxpasta error.

Source code in src/taxpasta/application/error/taxpasta_error.py 
25
26
27
def __int__(self, **kwargs) -> None:
    """Initialize a base taxpasta error."""
    super().__init__(**kwargs)

sample_merging_application

Provide a sample merging application.

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

SampleMergingApplication

Define a sample merging application.

Source code in src/taxpasta/application/sample_merging_application.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
class SampleMergingApplication:
    """Define a sample merging application."""

    def __init__(
        self,
        *,
        profile_reader: Type[ProfileReader],
        profile_standardiser: Type[ProfileStandardisationService],
        taxonomy_service: Optional[TaxonomyService] = None,
        **kwargs: dict,
    ):
        """
        Initialize the application for a particular taxonomic profiler.

        Args:
            profile_reader: A profile reader for a specific taxonomic profile format.
            profile_standardiser: A profile standardisation service for a specific
                taxonomic profile format.
            taxonomy_service: A taxonomy service instance. It is assumed that all
                profiles to be handled in the application are based on the given
                taxonomy loaded in the service instance.
            **kwargs: Passed on for inheritance.

        """
        super().__init__(**kwargs)
        self.reader = profile_reader
        self.standardiser = profile_standardiser
        self.taxonomy = taxonomy_service

    def run(
        self,
        profiles: Iterable[Tuple[str, Path]],
        wide_format: bool,
        summarise_at: Optional[str] = None,
        ignore_error: bool = False,
    ) -> DataFrame[WideObservationTable] | DataFrame[TidyObservationTable]:
        """
        Extract and transform profiles into samples, then merge them.

        Args:
            profiles: Pairs of name and profile path.
            wide_format: Whether to create wide or (tidy) long format output.
            summarise_at: The taxonomic rank at which to summarise abundance if any.
            ignore_error: Whether to ignore profiles that contain errors.

        Returns:
            A single table containing all samples in the desired format.

        Raises:
            StandardisationError: If any of the given profiles does not match the
                validation schema.  # noqa: DAR402

        """
        samples = self._etl_samples(profiles, ignore_error)

        if summarise_at is not None:
            samples = self._summarise_samples(samples, summarise_at, ignore_error)

        if wide_format:
            wide_table = SampleMergingService.merge_wide(samples)
            # If any profile did not have all the same taxonomy IDs as the combined
            # table, additional zeroes were introduced.
            if any(
                not wide_table[WideObservationTable.taxonomy_id]
                .isin(sample.profile[StandardProfile.taxonomy_id])
                .all()
                for sample in samples
            ):
                logger.warning(
                    "The merged profiles contained different taxa. Additional "
                    "zeroes were introduced for missing taxa."
                )
            return wide_table
        else:
            return SampleMergingService.merge_long(samples)

    def _etl_samples(
        self, profiles: Iterable[Tuple[str, Path]], ignore_error: bool
    ) -> List[Sample]:
        """Extract, transform, and load profiles into samples."""
        result = []
        for name, profile in profiles:
            try:
                result.append(
                    Sample(
                        name=name,
                        profile=self.standardiser.transform(self.reader.read(profile)),
                    )
                )
            except SchemaErrors as errors:
                if ignore_error:
                    logger.error("Sample %s: %s", name, str(errors))
                    continue
                else:
                    raise StandardisationError(
                        sample=name, profile=profile, message=str(errors.failure_cases)
                    ) from errors
            except ValueError as error:
                if ignore_error:
                    logger.error("Sample %s: %s", name, str(error))
                    continue
                else:
                    raise StandardisationError(
                        sample=name, profile=profile, message=str(error)
                    ) from error
        return result

    def _summarise_samples(
        self, samples: List[Sample], rank: str, ignore_error: bool
    ) -> List[Sample]:
        """Summarise samples at a given taxonomic rank."""
        assert self.taxonomy is not None  # nosec assert_used
        result = []
        for sample in samples:
            try:
                result.append(
                    Sample(
                        name=sample.name,
                        profile=self.taxonomy.summarise_at(sample.profile, rank),
                    )
                )
            except ValueError as error:
                if ignore_error:
                    logger.error("Sample %s: %s", sample.name, str(error))
                    continue
                else:
                    raise
        return result
Attributes
reader = profile_reader instance-attribute
standardiser = profile_standardiser instance-attribute
taxonomy = taxonomy_service instance-attribute
Functions
__init__(*, profile_reader: Type[ProfileReader], profile_standardiser: Type[ProfileStandardisationService], taxonomy_service: Optional[TaxonomyService] = None, **kwargs: dict)

Initialize the application for a particular taxonomic profiler.

Parameters:

Name Type Description Default
profile_reader Type[ProfileReader]

A profile reader for a specific taxonomic profile format.

 required
profile_standardiser Type[ProfileStandardisationService]

A profile standardisation service for a specific taxonomic profile format.

 required
taxonomy_service Optional[TaxonomyService]

A taxonomy service instance. It is assumed that all profiles to be handled in the application are based on the given taxonomy loaded in the service instance.

 None
**kwargs dict

Passed on for inheritance.

 {}
Source code in src/taxpasta/application/sample_merging_application.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def __init__(
    self,
    *,
    profile_reader: Type[ProfileReader],
    profile_standardiser: Type[ProfileStandardisationService],
    taxonomy_service: Optional[TaxonomyService] = None,
    **kwargs: dict,
):
    """
    Initialize the application for a particular taxonomic profiler.

    Args:
        profile_reader: A profile reader for a specific taxonomic profile format.
        profile_standardiser: A profile standardisation service for a specific
            taxonomic profile format.
        taxonomy_service: A taxonomy service instance. It is assumed that all
            profiles to be handled in the application are based on the given
            taxonomy loaded in the service instance.
        **kwargs: Passed on for inheritance.

    """
    super().__init__(**kwargs)
    self.reader = profile_reader
    self.standardiser = profile_standardiser
    self.taxonomy = taxonomy_service
run(profiles: Iterable[Tuple[str, Path]], wide_format: bool, summarise_at: Optional[str] = None, ignore_error: bool = False) -> DataFrame[WideObservationTable] | DataFrame[TidyObservationTable]

Extract and transform profiles into samples, then merge them.

Parameters:

Name Type Description Default
profiles Iterable[Tuple[str, Path]]

Pairs of name and profile path.

 required
wide_format bool

Whether to create wide or (tidy) long format output.

 required
summarise_at Optional[str]

The taxonomic rank at which to summarise abundance if any.

 None
ignore_error bool

Whether to ignore profiles that contain errors.

 False

Returns:

Type Description
DataFrame[WideObservationTable] | DataFrame[TidyObservationTable]

A single table containing all samples in the desired format.

Raises:

Type Description
StandardisationError

If any of the given profiles does not match the validation schema. # noqa: DAR402
Source code in src/taxpasta/application/sample_merging_application.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def run(
    self,
    profiles: Iterable[Tuple[str, Path]],
    wide_format: bool,
    summarise_at: Optional[str] = None,
    ignore_error: bool = False,
) -> DataFrame[WideObservationTable] | DataFrame[TidyObservationTable]:
    """
    Extract and transform profiles into samples, then merge them.

    Args:
        profiles: Pairs of name and profile path.
        wide_format: Whether to create wide or (tidy) long format output.
        summarise_at: The taxonomic rank at which to summarise abundance if any.
        ignore_error: Whether to ignore profiles that contain errors.

    Returns:
        A single table containing all samples in the desired format.

    Raises:
        StandardisationError: If any of the given profiles does not match the
            validation schema.  # noqa: DAR402

    """
    samples = self._etl_samples(profiles, ignore_error)

    if summarise_at is not None:
        samples = self._summarise_samples(samples, summarise_at, ignore_error)

    if wide_format:
        wide_table = SampleMergingService.merge_wide(samples)
        # If any profile did not have all the same taxonomy IDs as the combined
        # table, additional zeroes were introduced.
        if any(
            not wide_table[WideObservationTable.taxonomy_id]
            .isin(sample.profile[StandardProfile.taxonomy_id])
            .all()
            for sample in samples
        ):
            logger.warning(
                "The merged profiles contained different taxa. Additional "
                "zeroes were introduced for missing taxa."
            )
        return wide_table
    else:
        return SampleMergingService.merge_long(samples)

service

Attributes

Classes

Modules

profile_reader

Provide an abstract base class for reading taxonomic profiles.

Attributes
Classes
ProfileReader

Bases: ABC

Define an abstract base class for reading taxonomic profiles.

Source code in src/taxpasta/application/service/profile_reader.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class ProfileReader(ABC):
    """Define an abstract base class for reading taxonomic profiles."""

    @classmethod
    @abstractmethod
    def read(cls, profile: BufferOrFilepath) -> pd.DataFrame:
        """Read a taxonomic profile from the given source."""

    @classmethod
    def _check_num_columns(
        cls, profile: pd.DataFrame, schema_model: Type[pa.DataFrameModel]
    ) -> None:
        """Perform a strict test on the number of columns."""
        num_cols = len(schema_model.to_schema().columns)
        if len(profile.columns) != num_cols:
            raise ValueError(
                f"Unexpected report format. It has {len(profile.columns)} columns but "
                f"only {num_cols} are expected."
            )
Functions
read(profile: BufferOrFilepath) -> pd.DataFrame abstractmethod classmethod

Read a taxonomic profile from the given source.

Source code in src/taxpasta/application/service/profile_reader.py 
34
35
36
37
@classmethod
@abstractmethod
def read(cls, profile: BufferOrFilepath) -> pd.DataFrame:
    """Read a taxonomic profile from the given source."""
profile_standardisation_service

Provide an abstract base class for a profile standardisation service.

Classes
ProfileStandardisationService

Bases: ABC

Define an abstract base class for a profile standardisation service.

Source code in src/taxpasta/application/service/profile_standardisation_service.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class ProfileStandardisationService(ABC):
    """Define an abstract base class for a profile standardisation service."""

    @classmethod
    @abstractmethod
    def transform(cls, profile: DataFrame) -> DataFrame[StandardProfile]:
        """
        Tidy up and standardize a given taxonomic profile.

        Args:
            profile: The taxonomic profile of a particular tool.

        Returns:
            A standardized profile.

        """
Functions
transform(profile: DataFrame) -> DataFrame[StandardProfile] abstractmethod classmethod

Tidy up and standardize a given taxonomic profile.

Parameters:

Name Type Description Default
profile DataFrame

The taxonomic profile of a particular tool.

 required

Returns:

Type Description
DataFrame[StandardProfile]

A standardized profile.
Source code in src/taxpasta/application/service/profile_standardisation_service.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
@classmethod
@abstractmethod
def transform(cls, profile: DataFrame) -> DataFrame[StandardProfile]:
    """
    Tidy up and standardize a given taxonomic profile.

    Args:
        profile: The taxonomic profile of a particular tool.

    Returns:
        A standardized profile.

    """
standard_profile_writer

Provide an abstract base class for writing a standardized profile.

Attributes
Classes
StandardProfileWriter

Bases: ABC

Define an abstract base class for writing a standardized profile.

Source code in src/taxpasta/application/service/standard_profile_writer.py 
31
32
33
34
35
36
37
38
39
class StandardProfileWriter(ABC):
    """Define an abstract base class for writing a standardized profile."""

    @classmethod
    @abstractmethod
    def write(
        cls, profile: DataFrame[StandardProfile], target: BufferOrFilepath, **kwargs
    ) -> None:
        """Write a standardized profile to a file."""
Functions
write(profile: DataFrame[StandardProfile], target: BufferOrFilepath, **kwargs: BufferOrFilepath) -> None abstractmethod classmethod

Write a standardized profile to a file.

Source code in src/taxpasta/application/service/standard_profile_writer.py 
34
35
36
37
38
39
@classmethod
@abstractmethod
def write(
    cls, profile: DataFrame[StandardProfile], target: BufferOrFilepath, **kwargs
) -> None:
    """Write a standardized profile to a file."""
table_reader

Provide an abstract base class for reading tables.

Attributes
Classes
TableReader

Bases: ABC

Define an abstract base class for reading tables.

Source code in src/taxpasta/application/service/table_reader.py 
29
30
31
32
33
34
35
class TableReader(ABC):
    """Define an abstract base class for reading tables."""

    @classmethod
    @abstractmethod
    def read(cls, source: BufferOrFilepath, **kwargs) -> pd.DataFrame:
        """Read a table from the given source."""
Functions
read(source: BufferOrFilepath, **kwargs: BufferOrFilepath) -> pd.DataFrame abstractmethod classmethod

Read a table from the given source.

Source code in src/taxpasta/application/service/table_reader.py 
32
33
34
35
@classmethod
@abstractmethod
def read(cls, source: BufferOrFilepath, **kwargs) -> pd.DataFrame:
    """Read a table from the given source."""
tidy_observation_table_writer

Provide an abstract base class for writing tidy observation tables.

Attributes
Classes
TidyObservationTableWriter

Bases: ABC

Define an abstract base class for writing tidy observation tables.

Source code in src/taxpasta/application/service/tidy_observation_table_writer.py 
31
32
33
34
35
36
37
38
39
class TidyObservationTableWriter(ABC):
    """Define an abstract base class for writing tidy observation tables."""

    @classmethod
    @abstractmethod
    def write(
        cls, table: DataFrame[TidyObservationTable], target: BufferOrFilepath, **kwargs
    ) -> None:
        """Write a tidy observation table to the given buffer or file."""
Functions
write(table: DataFrame[TidyObservationTable], target: BufferOrFilepath, **kwargs: BufferOrFilepath) -> None abstractmethod classmethod

Write a tidy observation table to the given buffer or file.

Source code in src/taxpasta/application/service/tidy_observation_table_writer.py 
34
35
36
37
38
39
@classmethod
@abstractmethod
def write(
    cls, table: DataFrame[TidyObservationTable], target: BufferOrFilepath, **kwargs
) -> None:
    """Write a tidy observation table to the given buffer or file."""
wide_observation_table_writer

Provide an abstract base class for writing observation matrices.

Attributes
Classes
WideObservationTableWriter

Bases: ABC

Define an abstract base class for writing observation matrices.

Source code in src/taxpasta/application/service/wide_observation_table_writer.py 
31
32
33
34
35
36
37
38
39
class WideObservationTableWriter(ABC):
    """Define an abstract base class for writing observation matrices."""

    @classmethod
    @abstractmethod
    def write(
        cls, matrix: DataFrame[WideObservationTable], target: BufferOrFilepath, **kwargs
    ) -> None:
        """Write an observation matrix to the given buffer or file."""
Functions
write(matrix: DataFrame[WideObservationTable], target: BufferOrFilepath, **kwargs: BufferOrFilepath) -> None abstractmethod classmethod

Write an observation matrix to the given buffer or file.

Source code in src/taxpasta/application/service/wide_observation_table_writer.py 
34
35
36
37
38
39
@classmethod
@abstractmethod
def write(
    cls, matrix: DataFrame[WideObservationTable], target: BufferOrFilepath, **kwargs
) -> None:
    """Write an observation matrix to the given buffer or file."""