Skip to content

kmer

pykmhelpers.core.kmer

Kmer

Bases: Sequence

A k-mer sequence with a fixed length k.

Source code in pykmhelpers/core/kmer.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class Kmer(Sequence):
    """A k-mer sequence with a fixed length k."""

    def __init__(self, seq: str, k: int = 0, header: str = ""):
        if k == 0:
            k = len(seq)
        if len(seq) != k:
            raise ValueError(f"Sequence length {len(seq)} doesn't match k={k}")
        super().__init__(content=seq, header=header)
        self._k = k

    @property
    def k(self):
        return self._k

KmerCountMode

Bases: Enum

Counting strategy returned by KmerCounter.

DISTINCT: all distinct k-mers; suited for assembled sequences. SOLID: distinct k-mers appearing at least twice; filters sequencing errors from raw reads. TOTAL: total k-mer occurrences across all reads.

Source code in pykmhelpers/core/kmer.py 
26
27
28
29
30
31
32
33
34
35
36
37
class KmerCountMode(Enum):
    """Counting strategy returned by KmerCounter.

    DISTINCT: all distinct k-mers; suited for assembled sequences.
    SOLID:    distinct k-mers appearing at least twice;
              filters sequencing errors from raw reads.
    TOTAL:    total k-mer occurrences across all reads.
    """

    DISTINCT = 0
    SOLID = 1
    TOTAL = 2

KmerCounter

Bases: Wrapper

Wrapper around ntcard for counting k-mers in sequence files.

Parameters:

Name Type Description Default
k int

K-mer length.

 31
thread_count int

Number of threads passed to ntcard.

 8
dry_run bool

If True, commands are built but not executed.

 False
Source code in pykmhelpers/core/kmer.py 
 42
 43
 44
 45
 46
 47
 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
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
class KmerCounter(Wrapper):
    """Wrapper around ntcard for counting k-mers in sequence files.

    Args:
        k: K-mer length.
        thread_count: Number of threads passed to ntcard.
        dry_run: If True, commands are built but not executed.
    """

    def __init__(
        self,
        k: int = 31,
        thread_count: int = 8,
        dry_run: bool = False,
    ):
        super().__init__(main_cmd="ntcard", dry_run=dry_run)
        self._k = k
        self._thread_count = thread_count

    @property
    def k(self):
        """K-mer length."""
        return self._k

    @property
    def thread_count(self):
        """Number of threads used by ntcard."""
        return self._thread_count

    def count(
        self,
        filename: str,
        mode: KmerCountMode = KmerCountMode.DISTINCT,
        verbose: bool = False,
    ) -> int:
        """Count k-mers in a single file using ntcard.

        Args:
            filename: Path to the sequence file.
            mode: Counting strategy (see KmerCountMode).
            verbose: Print the mode name and resulting value to stdout.

        Returns:
            int: K-mer count extracted from the ntcard histogram according to mode.

        Raises:
            FileNotFoundError: If the sequence file does not exist.
            ValueError: If ntcard produces no output or parsing fails.
            subprocess.SubprocessError: If ntcard returns a non-zero exit code.
        """
        if not os.path.exists(filename):
            raise FileNotFoundError(f"Sequence file not found: {filename}")

        return self.count_files([filename], mode=mode, verbose=verbose)

    def count_files(
        self,
        files: list,
        mode: KmerCountMode = KmerCountMode.DISTINCT,
        verbose: bool = False,
    ) -> int:
        """Count k-mers for one or more files in a single ntcard call.

        All files are counted together as a single dataset; use count_all for
        per-sample counting.

        Args:
            files: List of sequence file paths.
            mode: Counting strategy (see KmerCountMode):
                  DISTINCT → F0 (all distinct k-mers),
                  SOLID    → F0 - freq[1] (distinct k-mers appearing at least twice),
                  TOTAL    → F1 (total k-mer occurrences).
            verbose: Print the mode name and resulting value to stdout.

        Returns:
            int: K-mer count extracted from the ntcard histogram according to mode.

        Raises:
            FileNotFoundError: If any input file does not exist.
            ValueError: If ntcard produces no output or parsing fails.
            subprocess.SubprocessError: If ntcard returns a non-zero exit code.
        """
        if not files:
            raise ValueError("At least one file is required")

        for file_path in files:
            if not os.path.exists(file_path):
                raise FileNotFoundError(f"File not found: {file_path}")

        with tempfile.NamedTemporaryFile(mode="w", delete=False) as tmp:
            tmp_file = tmp.name

        try:
            cmd = [
                "ntcard",
                "-t",
                str(self._thread_count),
                "-k",
                str(self._k),
                "-o",
                tmp_file,
            ] + files

            result = self._run_cmd(cmd, print_trace=verbose)

            hist_file = f"{tmp_file}_k{self._k}.hist"

            if os.path.exists(hist_file):
                # Old ntcard: "<prefix>_k<k>.hist" with "F1 <n>\nF0 <n>\n1 <n>\n..."
                with open(hist_file) as f:
                    lines = f.read().strip().split("\n")

                if len(lines) < 2:
                    raise ValueError("Unexpected ntcard output format")

                def parse_hist_line(line):
                    parts = line.split()
                    if len(parts) < 2:
                        raise ValueError(f"Cannot parse ntcard output line: {line!r}")
                    return int(parts[1])

                if mode == KmerCountMode.DISTINCT:
                    value = parse_hist_line(lines[1])  # F0
                elif mode == KmerCountMode.SOLID:
                    freq1 = parse_hist_line(lines[2]) if len(lines) > 2 else 0
                    value = parse_hist_line(lines[1]) - freq1  # F0 - freq[1]
                else:  # TOTAL
                    value = parse_hist_line(lines[0])  # F1

            elif os.path.getsize(tmp_file) > 0:
                # New ntcard: output written directly to "-o" path as TSV "k\tf\tn"
                # F0/F1 summary also available on stderr as "k=<k>\t<stat>\t<value>"
                with open(tmp_file) as f:
                    lines = f.read().strip().split("\n")

                freq: dict[int, int] = {}
                for line in lines[1:]:  # skip header "k  f  n"
                    parts = line.split()
                    if len(parts) >= 3:
                        try:
                            freq[int(parts[1])] = int(parts[2])
                        except ValueError:
                            pass

                f0 = sum(freq.values())
                freq1 = freq.get(1, 0)

                f1: int | None = None
                for line in result.stderr.splitlines():
                    parts = line.split()
                    if (
                        len(parts) >= 3
                        and parts[0].startswith("k=")
                        and parts[1] == "F1"
                    ):
                        try:
                            f1 = int(parts[2])
                        except ValueError:
                            pass

                if mode == KmerCountMode.DISTINCT:
                    value = f0
                elif mode == KmerCountMode.SOLID:
                    value = f0 - freq1
                else:  # TOTAL
                    value = f1 or 0

            else:
                raise ValueError("ntcard produced no output")

            if verbose:
                print(f"{mode.name}={value}")

            if value == 0:
                raise ValueError(
                    f"No k-mers found. Check if files are empty or k={self._k} is larger than sequences."
                )

            return value

        finally:
            hist_file = f"{tmp_file}_k{self._k}.hist"
            for path in (tmp_file, hist_file):
                if os.path.exists(path):
                    os.remove(path)

    def count_all(
        self,
        samples_dict: dict,
        mode: KmerCountMode = KmerCountMode.DISTINCT,
        verbose: bool = False,
    ) -> dict:
        """Count k-mers for multiple samples.

        Samples that already contain a "kmer_count" key are skipped.

        Args:
            samples_dict: Dictionary mapping sample names to their metadata::

                {
                    "sample_name": {"files": [list of file paths], ...},
                    ...
                }

            mode: Counting strategy applied to every sample (see KmerCountMode).
            verbose: Forward verbosity to each count_files call.

        Returns:
            dict[str, int]: K-mer count per sample name.

        Raises:
            ValueError: If a sample entry is malformed or ntcard fails.
        """
        results = {}

        for sample_name, sample_data in samples_dict.items():
            if not isinstance(sample_data, dict) or "files" not in sample_data:
                raise ValueError(
                    f"Invalid sample format for '{sample_name}': must have 'files' key"
                )

            if "kmer_count" in sample_data:
                continue

            files = sample_data["files"]
            if not isinstance(files, list):
                raise ValueError(
                    f"Invalid files format for '{sample_name}': 'files' must be a list"
                )

            if not files:
                raise ValueError(f"No files specified for sample '{sample_name}'")

            try:
                kmer_count = self.count_files(files, mode=mode, verbose=verbose)
                results[sample_name] = kmer_count

            except (FileNotFoundError, ValueError, subprocess.SubprocessError) as e:
                raise ValueError(
                    f"Error counting k-mers for sample '{sample_name}': {e}"
                ) from e

        return results

k property

K-mer length.

thread_count property

Number of threads used by ntcard.

count(filename, mode=KmerCountMode.DISTINCT, verbose=False)

Count k-mers in a single file using ntcard.

Parameters:

Name Type Description Default
filename str

Path to the sequence file.

 required
mode KmerCountMode

Counting strategy (see KmerCountMode).

 DISTINCT
verbose bool

Print the mode name and resulting value to stdout.

 False

Returns:

Name Type Description
int int

K-mer count extracted from the ntcard histogram according to mode.

Raises:

Type Description
FileNotFoundError

If the sequence file does not exist.
ValueError

If ntcard produces no output or parsing fails.
SubprocessError

If ntcard returns a non-zero exit code.
Source code in pykmhelpers/core/kmer.py 
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
def count(
    self,
    filename: str,
    mode: KmerCountMode = KmerCountMode.DISTINCT,
    verbose: bool = False,
) -> int:
    """Count k-mers in a single file using ntcard.

    Args:
        filename: Path to the sequence file.
        mode: Counting strategy (see KmerCountMode).
        verbose: Print the mode name and resulting value to stdout.

    Returns:
        int: K-mer count extracted from the ntcard histogram according to mode.

    Raises:
        FileNotFoundError: If the sequence file does not exist.
        ValueError: If ntcard produces no output or parsing fails.
        subprocess.SubprocessError: If ntcard returns a non-zero exit code.
    """
    if not os.path.exists(filename):
        raise FileNotFoundError(f"Sequence file not found: {filename}")

    return self.count_files([filename], mode=mode, verbose=verbose)

count_files(files, mode=KmerCountMode.DISTINCT, verbose=False)

Count k-mers for one or more files in a single ntcard call.

All files are counted together as a single dataset; use count_all for per-sample counting.

Parameters:

Name Type Description Default
files list

List of sequence file paths.

 required
mode KmerCountMode

Counting strategy (see KmerCountMode): DISTINCT → F0 (all distinct k-mers), SOLID → F0 - freq[1] (distinct k-mers appearing at least twice), TOTAL → F1 (total k-mer occurrences).

 DISTINCT
verbose bool

Print the mode name and resulting value to stdout.

 False

Returns:

Name Type Description
int int

K-mer count extracted from the ntcard histogram according to mode.

Raises:

Type Description
FileNotFoundError

If any input file does not exist.
ValueError

If ntcard produces no output or parsing fails.
SubprocessError

If ntcard returns a non-zero exit code.
Source code in pykmhelpers/core/kmer.py 
 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
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def count_files(
    self,
    files: list,
    mode: KmerCountMode = KmerCountMode.DISTINCT,
    verbose: bool = False,
) -> int:
    """Count k-mers for one or more files in a single ntcard call.

    All files are counted together as a single dataset; use count_all for
    per-sample counting.

    Args:
        files: List of sequence file paths.
        mode: Counting strategy (see KmerCountMode):
              DISTINCT → F0 (all distinct k-mers),
              SOLID    → F0 - freq[1] (distinct k-mers appearing at least twice),
              TOTAL    → F1 (total k-mer occurrences).
        verbose: Print the mode name and resulting value to stdout.

    Returns:
        int: K-mer count extracted from the ntcard histogram according to mode.

    Raises:
        FileNotFoundError: If any input file does not exist.
        ValueError: If ntcard produces no output or parsing fails.
        subprocess.SubprocessError: If ntcard returns a non-zero exit code.
    """
    if not files:
        raise ValueError("At least one file is required")

    for file_path in files:
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"File not found: {file_path}")

    with tempfile.NamedTemporaryFile(mode="w", delete=False) as tmp:
        tmp_file = tmp.name

    try:
        cmd = [
            "ntcard",
            "-t",
            str(self._thread_count),
            "-k",
            str(self._k),
            "-o",
            tmp_file,
        ] + files

        result = self._run_cmd(cmd, print_trace=verbose)

        hist_file = f"{tmp_file}_k{self._k}.hist"

        if os.path.exists(hist_file):
            # Old ntcard: "<prefix>_k<k>.hist" with "F1 <n>\nF0 <n>\n1 <n>\n..."
            with open(hist_file) as f:
                lines = f.read().strip().split("\n")

            if len(lines) < 2:
                raise ValueError("Unexpected ntcard output format")

            def parse_hist_line(line):
                parts = line.split()
                if len(parts) < 2:
                    raise ValueError(f"Cannot parse ntcard output line: {line!r}")
                return int(parts[1])

            if mode == KmerCountMode.DISTINCT:
                value = parse_hist_line(lines[1])  # F0
            elif mode == KmerCountMode.SOLID:
                freq1 = parse_hist_line(lines[2]) if len(lines) > 2 else 0
                value = parse_hist_line(lines[1]) - freq1  # F0 - freq[1]
            else:  # TOTAL
                value = parse_hist_line(lines[0])  # F1

        elif os.path.getsize(tmp_file) > 0:
            # New ntcard: output written directly to "-o" path as TSV "k\tf\tn"
            # F0/F1 summary also available on stderr as "k=<k>\t<stat>\t<value>"
            with open(tmp_file) as f:
                lines = f.read().strip().split("\n")

            freq: dict[int, int] = {}
            for line in lines[1:]:  # skip header "k  f  n"
                parts = line.split()
                if len(parts) >= 3:
                    try:
                        freq[int(parts[1])] = int(parts[2])
                    except ValueError:
                        pass

            f0 = sum(freq.values())
            freq1 = freq.get(1, 0)

            f1: int | None = None
            for line in result.stderr.splitlines():
                parts = line.split()
                if (
                    len(parts) >= 3
                    and parts[0].startswith("k=")
                    and parts[1] == "F1"
                ):
                    try:
                        f1 = int(parts[2])
                    except ValueError:
                        pass

            if mode == KmerCountMode.DISTINCT:
                value = f0
            elif mode == KmerCountMode.SOLID:
                value = f0 - freq1
            else:  # TOTAL
                value = f1 or 0

        else:
            raise ValueError("ntcard produced no output")

        if verbose:
            print(f"{mode.name}={value}")

        if value == 0:
            raise ValueError(
                f"No k-mers found. Check if files are empty or k={self._k} is larger than sequences."
            )

        return value

    finally:
        hist_file = f"{tmp_file}_k{self._k}.hist"
        for path in (tmp_file, hist_file):
            if os.path.exists(path):
                os.remove(path)

count_all(samples_dict, mode=KmerCountMode.DISTINCT, verbose=False)

Count k-mers for multiple samples.

Samples that already contain a "kmer_count" key are skipped.

Parameters:

Name Type Description Default
samples_dict dict

Dictionary mapping sample names to their metadata::

{ "sample_name": {"files": [list of file paths], ...}, ... }

 required
mode KmerCountMode

Counting strategy applied to every sample (see KmerCountMode).

 DISTINCT
verbose bool

Forward verbosity to each count_files call.

 False

Returns:

Type Description
dict

dict[str, int]: K-mer count per sample name.

Raises:

Type Description
ValueError

If a sample entry is malformed or ntcard fails.
Source code in pykmhelpers/core/kmer.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def count_all(
    self,
    samples_dict: dict,
    mode: KmerCountMode = KmerCountMode.DISTINCT,
    verbose: bool = False,
) -> dict:
    """Count k-mers for multiple samples.

    Samples that already contain a "kmer_count" key are skipped.

    Args:
        samples_dict: Dictionary mapping sample names to their metadata::

            {
                "sample_name": {"files": [list of file paths], ...},
                ...
            }

        mode: Counting strategy applied to every sample (see KmerCountMode).
        verbose: Forward verbosity to each count_files call.

    Returns:
        dict[str, int]: K-mer count per sample name.

    Raises:
        ValueError: If a sample entry is malformed or ntcard fails.
    """
    results = {}

    for sample_name, sample_data in samples_dict.items():
        if not isinstance(sample_data, dict) or "files" not in sample_data:
            raise ValueError(
                f"Invalid sample format for '{sample_name}': must have 'files' key"
            )

        if "kmer_count" in sample_data:
            continue

        files = sample_data["files"]
        if not isinstance(files, list):
            raise ValueError(
                f"Invalid files format for '{sample_name}': 'files' must be a list"
            )

        if not files:
            raise ValueError(f"No files specified for sample '{sample_name}'")

        try:
            kmer_count = self.count_files(files, mode=mode, verbose=verbose)
            results[sample_name] = kmer_count

        except (FileNotFoundError, ValueError, subprocess.SubprocessError) as e:
            raise ValueError(
                f"Error counting k-mers for sample '{sample_name}': {e}"
            ) from e

    return results