Skip to content

samtools

Plugin for samtools progress parsing.

Classes

SamtoolsIndexPlugin

Bases: ToolProgressPlugin

Progress parser for samtools index.

samtools index doesn't provide progress, but we can detect completion.

Source code in snakesee/plugins/samtools.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class SamtoolsIndexPlugin(ToolProgressPlugin):
    """
    Progress parser for samtools index.

    samtools index doesn't provide progress, but we can detect completion.
    """

    @property
    def tool_name(self) -> str:
        return "samtools-index"

    @property
    def tool_patterns(self) -> list[str]:
        return ["samtools index"]

    def can_parse(self, rule_name: str, log_content: str) -> bool:
        """Check if this looks like samtools index."""
        return "index" in rule_name.lower() and (
            "sam" in rule_name.lower() or "bam" in rule_name.lower()
        )

    def parse_progress(self, log_content: str) -> ToolProgress | None:
        """samtools index doesn't report progress."""
        return None

Functions

can_parse 
can_parse(rule_name: str, log_content: str) -> bool

Check if this looks like samtools index.

Source code in snakesee/plugins/samtools.py 
71
72
73
74
75
def can_parse(self, rule_name: str, log_content: str) -> bool:
    """Check if this looks like samtools index."""
    return "index" in rule_name.lower() and (
        "sam" in rule_name.lower() or "bam" in rule_name.lower()
    )
parse_progress 
parse_progress(log_content: str) -> ToolProgress | None

samtools index doesn't report progress.

Source code in snakesee/plugins/samtools.py 
77
78
79
def parse_progress(self, log_content: str) -> ToolProgress | None:
    """samtools index doesn't report progress."""
    return None

SamtoolsSortPlugin

Bases: ToolProgressPlugin

Progress parser for samtools sort.

Parses samtools sort progress output which shows processed reads.

Example samtools sort output:: [bam_sort_core] merging from 4 files and 1 in-memory blocks... [bam_sort_core] read 1000000 records...

Source code in snakesee/plugins/samtools.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class SamtoolsSortPlugin(ToolProgressPlugin):
    """
    Progress parser for samtools sort.

    Parses samtools sort progress output which shows processed reads.

    Example samtools sort output::
        [bam_sort_core] merging from 4 files and 1 in-memory blocks...
        [bam_sort_core] read 1000000 records...
    """

    # Pattern: read 1000000 records
    RECORDS_PATTERN = re.compile(r"read\s+(\d+)\s+records")

    @property
    def tool_name(self) -> str:
        return "samtools-sort"

    @property
    def tool_patterns(self) -> list[str]:
        return ["samtools sort", "samtools view"]

    def can_parse(self, rule_name: str, log_content: str) -> bool:
        """Check if this looks like samtools sort output."""
        if "sort" in rule_name.lower() and "sam" in rule_name.lower():
            return True
        return "[bam_sort_core]" in log_content

    def parse_progress(self, log_content: str) -> ToolProgress | None:
        """Extract progress from samtools sort log output."""
        total_records = 0

        for match in self.RECORDS_PATTERN.finditer(log_content):
            records = int(match.group(1))
            if records > total_records:
                total_records = records

        if total_records > 0:
            return ToolProgress(
                items_processed=total_records,
                items_total=None,
                unit="records",
            )

        return None

Functions

can_parse 
can_parse(rule_name: str, log_content: str) -> bool

Check if this looks like samtools sort output.

Source code in snakesee/plugins/samtools.py 
31
32
33
34
35
def can_parse(self, rule_name: str, log_content: str) -> bool:
    """Check if this looks like samtools sort output."""
    if "sort" in rule_name.lower() and "sam" in rule_name.lower():
        return True
    return "[bam_sort_core]" in log_content
parse_progress 
parse_progress(log_content: str) -> ToolProgress | None

Extract progress from samtools sort log output.

Source code in snakesee/plugins/samtools.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def parse_progress(self, log_content: str) -> ToolProgress | None:
    """Extract progress from samtools sort log output."""
    total_records = 0

    for match in self.RECORDS_PATTERN.finditer(log_content):
        records = int(match.group(1))
        if records > total_records:
            total_records = records

    if total_records > 0:
        return ToolProgress(
            items_processed=total_records,
            items_total=None,
            unit="records",
        )

    return None