How to use cutadapt - 10 common examples

def _run(params, inpath1, inpath2=None, expected1=None, expected2=None, cores=1):
        assert not (inpath1 and inpath2 and expected1 and expected2)
        assert not (expected2 and not expected1)
        assert not (inpath2 and not inpath1)
        if type(params) is str:
            params = params.split()
        params += ["--interleaved", "--cores", str(cores), "--buffer-size=512"]
        tmp1 = str(tmpdir.join("out1-" + expected1))
        params += ["-o", tmp1]
        paths = [datapath(inpath1)]
        if inpath2:
            paths += [datapath(inpath2)]
        if expected2:
            tmp2 = str(tmpdir.join("out2-" + expected2))
            params += ["-p", tmp2]
            assert main(params + paths) is None
            assert_files_equal(cutpath(expected2), tmp2)
        else:
            assert main(params + paths) is None
        assert_files_equal(cutpath(expected1), tmp1)

def test_interleaved_neither_nor(tmpdir):
    """Option --interleaved used, but pairs of files given for input and output"""
    p1 = str(tmpdir.join("temp-paired.1.fastq"))
    p2 = str(tmpdir.join("temp-paired.2.fastq"))
    params = "-a XX --interleaved".split()
    params += ["-o", p1, "-p1", p2, "paired.1.fastq", "paired.2.fastq"]
    with pytest.raises(SystemExit):
        main(params)

import shutil
from textwrap import dedent
import pytest
from nose.tools import raises
from tempfile import mkdtemp
from cutadapt.seqio import (Sequence, FormatError,
	FastaReader, FastqReader, InterleavedSequenceReader,
	FastaWriter, FastqWriter, InterleavedSequenceWriter, open as openseq,
	sequence_names_match, find_fastq_record_end,
	read_paired_chunks, read_chunks_from_file)
from cutadapt._seqio import two_fastq_heads  # re-exported


# files tests/data/simple.fast{q,a}
simple_fastq = [
	Sequence("first_sequence", "SEQUENCE1", ":6;;8<=:<"),
	Sequence("second_sequence", "SEQUENCE2", "83

def test(self):
		expected = [
			(Sequence('read1/1 some text', 'TTATTTGTCTCCAGC', '##HHHHHHHHHHHHH'),
			Sequence('read1/2 other text', 'GCTGGAGACAAATAA', 'HHHHHHHHHHHHHHH')),
			(Sequence('read3/1', 'CCAACTTGATATTAATAACA', 'HHHHHHHHHHHHHHHHHHHH'),
			Sequence('read3/2', 'TGTTATTAATATCAAGTTGG', '#HHHHHHHHHHHHHHHHHHH')),
			(Sequence('read5', 'TTATTTGTCTCCAGC', '#####HHHHHHHHHH'),
			Sequence('read5', 'CAACAGGCCACATTAGACATATCGGATGGT', 'HHHHHHHH##HHHHHHHHHHHHHHHHHHHH')),
		]
		reads = list(InterleavedSequenceReader("tests/cut/interleaved.fastq"))
		for (r1, r2), (e1, e2) in zip(reads, expected):
			print(r1, r2, e1, e2)

		assert reads == expected
		with openseq("tests/cut/interleaved.fastq", interleaved=True) as f:
			reads = list(f)
		assert reads == expected

def test(self):
		reads = [
			(Sequence('A/1 comment', 'TTA', '##H'),
			Sequence('A/2 comment', 'GCT', 'HH#')),
			(Sequence('B/1', 'CC', 'HH'),
			Sequence('B/2', 'TG', '#H'))
		]
		sio = StringIO()
		with InterleavedSequenceWriter(sio) as writer:
			for read1, read2 in reads:
				writer.write(read1, read2)
		assert sio.getvalue() == '@A/1 comment\nTTA\n+\n##H\n@A/2 comment\nGCT\n+\nHH#\n@B/1\nCC\n+\nHH\n@B/2\nTG\n+\n#H\n'

def test(self):
		reads = [
			(Sequence('A/1 comment', 'TTA', '##H'),
			Sequence('A/2 comment', 'GCT', 'HH#')),
			(Sequence('B/1', 'CC', 'HH'),
			Sequence('B/2', 'TG', '#H'))
		]
		sio = StringIO()
		with InterleavedSequenceWriter(sio) as writer:
			for read1, read2 in reads:
				writer.write(read1, read2)
		assert sio.getvalue() == '@A/1 comment\nTTA\n+\n##H\n@A/2 comment\nGCT\n+\nHH#\n@B/1\nCC\n+\nHH\n@B/2\nTG\n+\n#H\n'

def test(self):
		expected = [
			(Sequence('read1/1 some text', 'TTATTTGTCTCCAGC', '##HHHHHHHHHHHHH'),
			Sequence('read1/2 other text', 'GCTGGAGACAAATAA', 'HHHHHHHHHHHHHHH')),
			(Sequence('read3/1', 'CCAACTTGATATTAATAACA', 'HHHHHHHHHHHHHHHHHHHH'),
			Sequence('read3/2', 'TGTTATTAATATCAAGTTGG', '#HHHHHHHHHHHHHHHHHHH')),
			(Sequence('read5', 'TTATTTGTCTCCAGC', '#####HHHHHHHHHH'),
			Sequence('read5', 'CAACAGGCCACATTAGACATATCGGATGGT', 'HHHHHHHH##HHHHHHHHHHHHHHHHHHHH')),
		]
		reads = list(InterleavedSequenceReader("tests/cut/interleaved.fastq"))
		for (r1, r2), (e1, e2) in zip(reads, expected):
			print(r1, r2, e1, e2)

		assert reads == expected
		with openseq("tests/cut/interleaved.fastq", interleaved=True) as f:
			reads = list(f)
		assert reads == expected

def test(self):
		with FastaReader("tests/data/simple.fasta") as f:
			reads = list(f)
		assert reads == simple_fasta

		fasta = StringIO(">first_sequence\nSEQUENCE1\n>second_sequence\nSEQUENCE2\n")
		reads = list(FastaReader(fasta))
		assert reads == simple_fasta

def test(self):
		with FastaReader("tests/data/simple.fasta") as f:
			reads = list(f)
		assert reads == simple_fasta

		fasta = StringIO(">first_sequence\nSEQUENCE1\n>second_sequence\nSEQUENCE2\n")
		reads = list(FastaReader(fasta))
		assert reads == simple_fasta

def test(self):
		with FastaWriter(self.path) as fw:
			fw.write("name", "CCATA")
			fw.write("name2", "HELLO")
		assert fw._file.closed
		with open(self.path) as t:
			assert t.read() == '>name\nCCATA\n>name2\nHELLO\n'