class Progress:
    """
    Print an animated progress report to sys.stderr
    """

    def __init__(self, every=1):
        """
        every: minimum time to wait in seconds between progress updates
        """
        self._every = every
        self._n = 0
        # Time at which the progress was last updated
        self._time = time.time()
        self._start_time = self._time
        self._animation = self.scissors()

    def __repr__(self):
        return f"Progress(_n={self._n}, elapsed={self._time - self._start_time:.3f})"

    @staticmethod
    def scissors(width=10):
        while True:
            for is_reverse, rang in [
                (False, range(width + 1)),
                (True, range(width + 1)),
            ]:
                for position in rang:
                    for is_open in (True, False):
                        left = " " * position
                        right = "-" * (width - position)
                        if is_reverse:
                            sc = ">8" if is_open else "=8"
                            left, right = right, left
                        else:
                            sc = "8<" if is_open else "8="
                        yield "[" + left + sc + right + "]"

    def update(self, total, _final=False):
        current_time = time.time()
        if _final:
            time_delta = current_time - self._start_time
            delta = total
        else:
            time_delta = current_time - self._time
            delta = total - self._n
        if delta < 1:
            return
        if time_delta == 0:
            return
        if not _final:
            if time_delta < self._every:
                return

        t = current_time - self._start_time
        hours = int(t) // 3600
        minutes = (int(t) - hours * 3600) // 60
        seconds = int(t) % 60
        per_second = delta / time_delta
        per_item = time_delta / delta

        animation = next(self._animation)
        if _final:
            animation = "Done".ljust(len(animation))
        print(
            "\r"
            "{animation} {hours:02d}:{minutes:02d}:{seconds:02d} "
            "{total:13,d} reads @ {per_item:5.1F} µs/read; {per_minute:6.2F} M reads/minute"
            "".format(
                hours=hours,
                minutes=minutes,
                seconds=seconds,
                total=total,
                per_item=per_item * 1e6,
                per_minute=per_second * 60 / 1e6,
                animation=animation,
            ),
            end="",
            file=sys.stderr,
        )
        self._n = total
        self._time = current_time

    def stop(self, total):
        """
        Print final progress reflecting the final total
        """
        self.update(total, _final=True)
        print(file=sys.stderr) class FileOpener:
    def __init__(self, compression_level: int = 6, threads: int = None):
        """
        threads -- no. of external compression threads.
            0: write in-process
            None: min(cpu_count(0, 4)
        """
        self.compression_level = compression_level
        self.threads = threads

    def xopen(self, path, mode):
        threads = self.threads if "w" in mode else 0
        f = open_raise_limit(
            xopen, path, mode, compresslevel=self.compression_level, threads=threads
        )
        logger.debug("Opening '%s', mode '%s' with xopen resulted in %s", path, mode, f)
        return f

    def xopen_or_none(self, path, mode):
        """Return opened file or None if the path is None"""
        if path is None:
            return None
        return self.xopen(path, mode)

    def xopen_pair(self, path1: str, path2: Optional[str], mode):
        if path1 is None and path2 is not None:
            raise ValueError(
                "When giving paths for paired-end files, only providing the second"
                " file is not supported"
            )
        file1 = self.xopen_or_none(path1, mode)
        file2 = self.xopen_or_none(path2, mode)
        return file1, file2

    def dnaio_open(self, *args, **kwargs):
        kwargs["opener"] = self.xopen
        f = dnaio.open(*args, **kwargs)
        logger.debug(
            "Opening %r, mode '%s' with dnaio resulted in %s",
            args[0],
            kwargs["mode"],
            f,
        )
        return f

    def dnaio_open_raise_limit(self, *args, **kwargs):
        """
        Open a FASTA/FASTQ file for writing. If it fails because the number of open files
        would be exceeded, try to raise the soft limit and re-try.
        """
        return open_raise_limit(self.dnaio_open, *args, **kwargs) class SingleEndModifier(ABC):
    @abstractmethod
    def __call__(self, read, info: ModificationInfo):
        pass class PairedEndModifier(ABC):
    @abstractmethod
    def __call__(
        self, read1, read2, info1: ModificationInfo, info2: ModificationInfo
    ) -> Tuple[SequenceRecord, SequenceRecord]:
        pass class PairedEndModifierWrapper(PairedEndModifier):
    """
    Wrapper for modifiers that work on both reads in a paired-end read
    """

    paired = True

    def __init__(
        self,
        modifier1: Optional[SingleEndModifier],
        modifier2: Optional[SingleEndModifier],
    ):
        """Set one of the modifiers to None to work on R1 or R2 only"""
        self._modifier1 = modifier1
        self._modifier2 = modifier2
        if self._modifier1 is None and self._modifier2 is None:
            raise ValueError("Not both modifiers may be None")

    def __repr__(self):
        return (
            "PairedEndModifierWrapper("
            f"modifier1={self._modifier1!r}, modifier2={self._modifier2!r})"
        )

    def __call__(self, read1, read2, info1: ModificationInfo, info2: ModificationInfo):
        if self._modifier1 is None:
            return read1, self._modifier2(read2, info2)  # type: ignore
        if self._modifier2 is None:
            return self._modifier1(read1, info1), read2
        return self._modifier1(read1, info1), self._modifier2(read2, info2) class ModificationInfo:
    """
    An object of this class is created for each read that passes through the pipeline.
    Any information (except the read itself) that needs to be passed from one modifier
    to one later in the pipeline or from one modifier to the filters is recorded here.
    """

    __slots__ = ["matches", "original_read", "cut_prefix", "cut_suffix", "is_rc"]

    def __init__(self, read):
        self.matches = []  # type: List[Match]
        self.original_read = read
        self.cut_prefix = None
        self.cut_suffix = None
        self.is_rc = None

    def __repr__(self):
        return (
            "ModificationInfo("
            f"matches={self.matches!r}, "
            f"original_read={self.original_read}, "
            f"cut_prefix={self.cut_prefix}, "
            f"cut_suffix={self.cut_suffix}, "
            f"is_rc={self.is_rc})"
        ) class Statistics:
    def __init__(self) -> None:
        """ """
        self.paired: Optional[bool] = None
        # Map a filter name to the number of filtered reads/read pairs
        self.filtered: Dict[str, int] = defaultdict(int)
        self.reverse_complemented: Optional[int] = None
        self.n = 0
        self.total_bp = [0, 0]
        self.read_length_statistics = ReadLengthStatistics()
        self.with_adapters: List[Optional[int]] = [None, None]
        self.quality_trimmed_bp: List[Optional[int]] = [None, None]
        self.adapter_stats: List[List[AdapterStatistics]] = [[], []]
        self._collected: bool = False

    def __iadd__(self, other: Any):
        if not isinstance(other, Statistics):
            raise ValueError(f"Cannot add {other.__type__.__name__}")
        self.n += other.n
        self.read_length_statistics += other.read_length_statistics

        if self.paired is None:
            self.paired = other.paired
        elif self.paired != other.paired:
            raise ValueError("Incompatible Statistics: paired is not equal")

        self.reverse_complemented = add_if_not_none(
            self.reverse_complemented, other.reverse_complemented
        )

        for filter_name, count in other.filtered.items():
            self.filtered[filter_name] += count

        for i in (0, 1):
            self.total_bp[i] += other.total_bp[i]
            self.with_adapters[i] = add_if_not_none(
                self.with_adapters[i], other.with_adapters[i]
            )
            self.quality_trimmed_bp[i] = add_if_not_none(
                self.quality_trimmed_bp[i], other.quality_trimmed_bp[i]
            )
            if self.adapter_stats[i] and other.adapter_stats[i]:
                if len(self.adapter_stats[i]) != len(other.adapter_stats[i]):
                    raise ValueError(
                        "Incompatible Statistics objects (adapter_stats length)"
                    )
                for j in range(len(self.adapter_stats[i])):
                    self.adapter_stats[i][j] += other.adapter_stats[i][j]
            elif other.adapter_stats[i]:
                assert self.adapter_stats[i] == []
                self.adapter_stats[i] = other.adapter_stats[i]
        return self

    def collect(
        self, n: int, total_bp1: int, total_bp2: Optional[int], modifiers, steps
    ):
        """
        n -- total number of reads
        total_bp1 -- number of bases in first reads
        total_bp2 -- number of bases in second reads. None for single-end data.
        """
        if self._collected:
            raise ValueError("Cannot call Statistics.collect more than once")
        self.n = n
        self.total_bp[0] = total_bp1
        if total_bp2 is None:
            self.paired = False
        else:
            self.paired = True
            self.total_bp[1] = total_bp2

        for step in steps:
            self._collect_step(step)
        for modifier in modifiers:
            self._collect_modifier(modifier)
        self._collected = True

        # For chaining
        return self

    def _collect_step(self, step) -> None:
        if isinstance(step, HasStatistics):
            self.read_length_statistics += step.get_statistics()
        if isinstance(step, (SingleEndFilter, PairedEndFilter)):
            name = step.descriptive_identifier()
            self.filtered[name] = step.filtered

    def _collect_modifier(self, m) -> None:
        if isinstance(m, PairedAdapterCutter):
            for i in 0, 1:
                self.with_adapters[i] = m.with_adapters
                self.adapter_stats[i] = list(m.adapter_statistics[i].values())
            return
        if isinstance(m, PairedEndModifierWrapper):
            modifiers_list = [(0, m._modifier1), (1, m._modifier2)]
        else:
            modifiers_list = [(0, m)]
        for i, modifier in modifiers_list:
            if isinstance(modifier, (QualityTrimmer, NextseqQualityTrimmer)):
                self.quality_trimmed_bp[i] = modifier.trimmed_bases
            elif isinstance(modifier, AdapterCutter):
                assert self.with_adapters[i] is None
                self.with_adapters[i] = modifier.with_adapters
                self.adapter_stats[i] = list(modifier.adapter_statistics.values())
            elif isinstance(modifier, ReverseComplementer):
                assert self.with_adapters[i] is None
                self.with_adapters[i] = modifier.adapter_cutter.with_adapters
                self.adapter_stats[i] = list(
                    modifier.adapter_cutter.adapter_statistics.values()
                )
                self.reverse_complemented = modifier.reverse_complemented

    def as_json(self, gc_content: float) -> Dict:
        """Return a dict representation suitable for dumping in JSON format"""
        filtered = {name: self.filtered.get(name) for name in FILTERS.keys()}
        filtered_total = sum(self.filtered.values())
        written_reads = self.read_length_statistics.written_reads()
        written_bp = self.read_length_statistics.written_bp()
        assert written_reads + filtered_total == self.n
        return {
            "read_counts": {  # pairs or reads
                "input": self.n,
                "filtered": filtered,
                "output": self.read_length_statistics.written_reads(),
                "reverse_complemented": self.reverse_complemented,
                "read1_with_adapter": self.with_adapters[0],
                "read2_with_adapter": self.with_adapters[1] if self.paired else None,
            },
            "basepair_counts": {
                "input": self.total,
                "input_read1": self.total_bp[0],
                "input_read2": self.total_bp[1] if self.paired else None,
                "quality_trimmed": self.quality_trimmed,
                "quality_trimmed_read1": self.quality_trimmed_bp[0],
                "quality_trimmed_read2": self.quality_trimmed_bp[1],
                "output": self.total_written_bp,
                "output_read1": written_bp[0],
                "output_read2": written_bp[1] if self.paired else None,
            },
            "adapters_read1": [
                self._adapter_statistics_as_json(astats, self.n, gc_content)
                for astats in self.adapter_stats[0]
            ],
            "adapters_read2": [
                self._adapter_statistics_as_json(astats, self.n, gc_content)
                for astats in self.adapter_stats[1]
            ]
            if self.paired
            else None,
        }

    def _adapter_statistics_as_json(
        self, adapter_statistics: AdapterStatistics, n: int, gc_content: float
    ):
        adapter = adapter_statistics.adapter
        ends: List[Optional[Dict[str, Any]]] = []
        total_trimmed_reads = 0
        for end_statistics in adapter_statistics.end_statistics():
            if end_statistics is None:
                ends.append(None)
                continue
            total = sum(end_statistics.lengths.values())
            if end_statistics.allows_partial_matches:
                eranges = ErrorRanges(
                    length=end_statistics.effective_length,
                    error_rate=end_statistics.max_error_rate,
                ).lengths()
            else:
                eranges = None
            base_stats = AdjacentBaseStatistics(end_statistics.adjacent_bases)
            trimmed_lengths = [
                OneLine(
                    {
                        "len": row.length,
                        "expect": round(row.expect, 1),
                        "counts": row.error_counts,
                    }
                )
                for row in histogram_rows(end_statistics, n, gc_content)
            ]
            ends.append(
                {
                    "type": end_statistics.adapter_type,
                    "sequence": end_statistics.sequence,
                    "error_rate": end_statistics.max_error_rate,
                    "indels": end_statistics.indels,
                    "error_lengths": OneLine(eranges),
                    "matches": total,
                    "adjacent_bases": base_stats.as_json(),
                    "dominant_adjacent_base": base_stats.warnbase,
                    "trimmed_lengths": trimmed_lengths,
                }
            )
            total_trimmed_reads += total

        on_reverse_complement = (
            adapter_statistics.reverse_complemented
            if self.reverse_complemented
            else None
        )
        return {
            "name": adapter_statistics.name,
            "total_matches": total_trimmed_reads,
            "on_reverse_complement": on_reverse_complement,
            "linked": isinstance(adapter, LinkedAdapter),
            "five_prime_end": ends[0],
            "three_prime_end": ends[1],
        }

    @property
    def total(self) -> int:
        return sum(self.total_bp)

    @property
    def quality_trimmed(self) -> Optional[int]:
        return add_if_not_none(*self.quality_trimmed_bp)

    @property
    def total_written_bp(self) -> int:
        return sum(self.read_length_statistics.written_bp())

    @property
    def written(self) -> int:
        return self.read_length_statistics.written_reads()

    @property
    def written_fraction(self) -> float:
        return safe_divide(self.read_length_statistics.written_reads(), self.n)

    @property
    def with_adapters_fraction(self) -> List[float]:
        return [safe_divide(v, self.n) for v in self.with_adapters]

    @property
    def quality_trimmed_fraction(self) -> float:
        return safe_divide(self.quality_trimmed, self.total)

    @property
    def written_bp(self) -> Tuple[int, int]:
        return self.read_length_statistics.written_bp()

    @property
    def total_written_bp_fraction(self) -> float:
        return safe_divide(self.total_written_bp, self.total)

    @property
    def reverse_complemented_fraction(self) -> float:
        return safe_divide(self.reverse_complemented, self.n)

    def filtered_fraction(self, filter_name: str) -> float:
        return safe_divide(self.filtered.get(filter_name), self.n) class TooShort(Predicate):
    """Select reads that are shorter than the specified minimum length"""

    def __init__(self, minimum_length: int):
        self.minimum_length = minimum_length

    def __repr__(self):
        return f"TooShort(minimum_length={self.minimum_length})"

    def test(self, read, info: ModificationInfo):
        return len(read) < self.minimum_length class TooLong(Predicate):
    """Select reads that are longer than the specified maximum length"""

    def __init__(self, maximum_length: int):
        self.maximum_length = maximum_length

    def __repr__(self):
        return f"TooLong(maximum_length={self.maximum_length})"

    def test(self, read, info: ModificationInfo):
        return len(read) > self.maximum_length class TooManyN(Predicate):
    """
    Select reads that have too many 'N' bases.

    Both a raw count or a proportion (relative to the sequence length) can be used.
    """

    def __init__(self, count: float):
        """
        count -- if it is below 1.0, it will be considered a proportion, and above and equal to
        1 will be considered as discarding reads with a number of N's greater than this cutoff.
        """
        assert count >= 0
        self.is_proportion = count < 1.0
        self.cutoff = count

    def __repr__(self):
        return f"TooManyN(cutoff={self.cutoff}, is_proportion={self.is_proportion})"

    def test(self, read, info: ModificationInfo):
        n_count = read.sequence.lower().count("n")
        if self.is_proportion:
            if len(read) == 0:
                return False
            return n_count / len(read) > self.cutoff
        else:
            return n_count > self.cutoff class TooManyExpectedErrors(Predicate):
    """
    Select reads whose expected number of errors, according to the quality
    values, exceeds a threshold.

    The idea comes from usearch's -fastq_maxee parameter
    (http://drive5.com/usearch/).
    """

    def __init__(self, max_errors: float):
        self.max_errors = max_errors

    def __repr__(self):
        return f"TooManyExpectedErrors(max_errors={self.max_errors})"

    def test(self, read, info: ModificationInfo):
        return expected_errors(read.qualities) > self.max_errors class CasavaFiltered(Predicate):
    """
    Select reads that have failed the CASAVA filter according to the read header.
    The headers look like ``xxxx x:Y:x:x`` (with a ``Y``). Reads that pass the filter
    have an ``N`` instead of ``Y``.

    Reads with unrecognized headers are not selected.
    """

    def __repr__(self):
        return "CasavaFiltered()"

    def test(self, read, info: ModificationInfo):
        _, _, right = read.name.partition(" ")
        return right[1:4] == ":Y:"  # discard if :Y: found class DiscardTrimmed(Predicate):
    """
    Select reads for which at least one adapter match was found
    """

    def __repr__(self):
        return "DiscardTrimmed()"

    def test(self, read, info: ModificationInfo):
        return bool(info.matches) class DiscardUntrimmed(Predicate):
    """
    Select reads for which no adapter match was found
    """

    def __repr__(self):
        return "DiscardUntrimmed()"

    def test(self, read, info: ModificationInfo):
        return not info.matches class Predicate(ABC):
    @abstractmethod
    def test(self, read, info: ModificationInfo) -> bool:
        """
        Return True if the filtering criterion matches.
        """

    @classmethod
    def descriptive_identifier(cls) -> str:
        """
        Return a short name for this predicate based on the class name such as "too_long",
        "too_many_expected_errors".
        This is used as identifier in the JSON report.
        """
        return "".join(
            ("_" + ch.lower() if ch.isupper() else ch) for ch in cls.__name__
        )[1:] class SingleEndSink(SingleEndStep, HasStatistics):
    """
    Send each read to a writer and keep read length statistics.
    This is used as the last step in a pipeline.
    """

    def __init__(self, writer):
        super().__init__()
        self.writer = writer
        self._statistics = ReadLengthStatistics()

    def __repr__(self):
        return f"NoFilter({self.writer})"

    def __call__(self, read, info: ModificationInfo) -> bool:
        self.writer.write(read)
        self._statistics.update(read)
        return DISCARD

    def get_statistics(self) -> ReadLengthStatistics:
        return self._statistics class PairedEndSink(PairedEndStep, HasStatistics):
    """
    Send each read pair to a writer and keep read length statistics.
    This is used as the last step in a pipeline.
    """

    def __init__(self, writer):
        super().__init__()
        self.writer = writer
        self._statistics = ReadLengthStatistics()

    def __repr__(self):
        return f"PairedNoFilter({self.writer})"

    def __call__(
        self, read1, read2, info1: ModificationInfo, info2: ModificationInfo
    ) -> bool:
        self.writer.write(read1, read2)
        self._statistics.update2(read1, read2)
        return DISCARD

    def get_statistics(self) -> ReadLengthStatistics:
        return self._statistics class SingleEndFilter(SingleEndStep):
    """
    A pipeline step that can filter reads, can redirect filtered ones to a writer, and
    counts how many were filtered.
    """

    def __init__(self, writer, predicate: Predicate):
        super().__init__()
        self.filtered = 0
        self.writer = writer
        self.predicate = predicate

    def __repr__(self):
        return f"SingleEndFilter(writer={self.writer}, predicate={self.predicate})"

    def descriptive_identifier(self) -> str:
        return self.predicate.descriptive_identifier()

    def __call__(self, read, info: ModificationInfo) -> bool:
        if self.predicate.test(read, info):
            self.filtered += 1
            if self.writer is not None:
                self.writer.write(read)
            return DISCARD
        return KEEP class PairedEndFilter(PairedEndStep):
    """
    A pipeline step that can filter paired-end reads, redirect them to a file, and counts
    how many read pairs were filtered.

    Different filtering styles are supported, differing by which of the
    two reads in a pair have to fulfill the filtering criterion.
    """

    def __init__(
        self,
        writer,
        predicate1: Predicate,
        predicate2: Predicate,
        pair_filter_mode="any",
    ):
        """
        pair_filter_mode -- these values are allowed:
            'any': The pair is discarded if any read matches.
            'both': The pair is discarded if both reads match.
            'first': The pair is discarded if the first read matches.
        """
        super().__init__()
        if pair_filter_mode not in ("any", "both", "first"):
            raise ValueError("pair_filter_mode must be 'any', 'both' or 'first'")
        self._pair_filter_mode = pair_filter_mode
        self.filtered = 0
        self.writer = writer
        self.predicate1 = predicate1
        self.predicate2 = predicate2
        self._is_filtered: Any
        if predicate2 is None:
            self._is_filtered = self._is_filtered_first
        elif predicate1 is None:
            self._is_filtered = self._is_filtered_second
        elif pair_filter_mode == "any":
            self._is_filtered = self._is_filtered_any
        elif pair_filter_mode == "both":
            self._is_filtered = self._is_filtered_both
        else:
            self._is_filtered = self._is_filtered_first

    def __repr__(self):
        return (
            f"PairedEndFilter(writer={self.writer}, predicate1={self.predicate1}, "
            f"predicate2={self.predicate2}, pair_filter_mode='{self._pair_filter_mode}')"
        )

    def descriptive_identifier(self) -> str:
        if self.predicate1 is not None:
            return self.predicate1.descriptive_identifier()
        else:
            return self.predicate2.descriptive_identifier()

    def _is_filtered_any(
        self, read1, read2, info1: ModificationInfo, info2: ModificationInfo
    ) -> bool:
        return self.predicate1.test(read1, info1) or self.predicate2.test(read2, info2)

    def _is_filtered_both(
        self, read1, read2, info1: ModificationInfo, info2: ModificationInfo
    ) -> bool:
        return self.predicate1.test(read1, info1) and self.predicate2.test(read2, info2)

    def _is_filtered_first(
        self, read1, read2, info1: ModificationInfo, info2: ModificationInfo
    ) -> bool:
        return self.predicate1.test(read1, info1)

    def _is_filtered_second(
        self, read1, read2, info1: ModificationInfo, info2: ModificationInfo
    ) -> bool:
        return self.predicate2.test(read2, info2)

    def __call__(
        self, read1, read2, info1: ModificationInfo, info2: ModificationInfo
    ) -> bool:
        if self._is_filtered(read1, read2, info1, info2):
            self.filtered += 1
            if self.writer is not None:
                self.writer.write(read1, read2)
            return DISCARD
        return KEEP class Demultiplexer(SingleEndStep, HasStatistics):
    """
    Demultiplex trimmed reads. Reads are written to different output files
    depending on which adapter matches.

    Untrimmed reads are sent to writers[None] if that key exists.
    """

    def __init__(self, writers: Dict[Optional[str], Any]):
        """
        writers maps an adapter name to a writer
        """
        super().__init__()
        self._writers = writers
        self._untrimmed_writer = self._writers.get(None, None)
        self._statistics = ReadLengthStatistics()

    def __repr__(self):
        return f"<Demultiplexer len(writers)={len(self._writers)}>"

    def __call__(self, read, info):
        """
        Write the read to the proper output file according to the most recent match
        """
        if info.matches:
            name = info.matches[-1].adapter.name
            self._statistics.update(read)
            self._writers[name].write(read)
        elif self._untrimmed_writer is not None:
            self._statistics.update(read)
            self._untrimmed_writer.write(read)
        return DISCARD

    def get_statistics(self) -> ReadLengthStatistics:
        return self._statistics class PairedDemultiplexer(PairedEndStep, HasStatistics):
    """
    Demultiplex trimmed paired-end reads. Reads are written to different output files
    depending on which adapter (in read 1) matches.
    """

    def __init__(self, writers: Dict[Optional[str], Any]):
        super().__init__()
        self._writers = writers
        self._untrimmed_writer = self._writers.get(None, None)
        self._statistics = ReadLengthStatistics()

    def __call__(
        self, read1, read2, info1: ModificationInfo, info2: ModificationInfo
    ) -> bool:
        assert read2 is not None
        if info1.matches:
            name = info1.matches[-1].adapter.name  # type: ignore
            self._statistics.update2(read1, read2)
            self._writers[name].write(read1, read2)
        elif self._untrimmed_writer is not None:
            self._statistics.update2(read1, read2)
            self._untrimmed_writer.write(read1, read2)
        return DISCARD

    def get_statistics(self) -> ReadLengthStatistics:
        return self._statistics class CombinatorialDemultiplexer(PairedEndStep, HasStatistics):
    """
    Demultiplex paired-end reads depending on which adapter matches, taking into account
    matches on R1 and R2.
    """

    def __init__(self, writers: Dict[Tuple[Optional[str], Optional[str]], Any]):
        """
        Adapter names of the matches on R1 and R2 will be used to look up the writer in the
        writers dict. If there is no match on a read, None is used in the lookup instead
        of the name. Missing dictionary keys are ignored and can be used to discard
        read pairs.
        """
        super().__init__()
        self._writers = writers
        self._statistics = ReadLengthStatistics()

    def __call__(self, read1, read2, info1, info2) -> bool:
        """
        Write the read to the proper output file according to the most recent matches both on
        R1 and R2
        """
        assert read2 is not None
        name1 = info1.matches[-1].adapter.name if info1.matches else None
        name2 = info2.matches[-1].adapter.name if info2.matches else None
        key = (name1, name2)
        if key in self._writers:
            self._statistics.update2(read1, read2)
            self._writers[key].write(read1, read2)
        return DISCARD

    def get_statistics(self) -> ReadLengthStatistics:
        return self._statistics class RestFileWriter(SingleEndStep):
    def __init__(self, file):
        self._file = file

    def __repr__(self):
        return f"RestFileWriter(file={self._file})"

    def __call__(self, read, info):
        # TODO this fails with linked adapters
        if info.matches:
            rest = info.matches[-1].rest()
            if len(rest) > 0:
                print(rest, read.name, file=self._file)
        return KEEP class WildcardFileWriter(SingleEndStep):
    def __init__(self, file):
        self._file = file

    def __repr__(self):
        return f"WildcardFileWriter(file={self._file})"

    def __call__(self, read, info) -> bool:
        # TODO this fails with linked adapters
        if info.matches:
            print(info.matches[-1].wildcards(), read.name, file=self._file)
        return KEEP class InfoFileWriter(SingleEndStep):
    RC_MAP = {None: "", True: "1", False: "0"}

    def __init__(self, file):
        self._file = file

    def __repr__(self):
        return f"InfoFileWriter(file={self._file})"

    def __call__(self, read, info: ModificationInfo):
        current_read = info.original_read
        if info.is_rc:
            current_read = reverse_complemented_sequence(current_read)
        if info.matches:
            for match in info.matches:
                for info_record in match.get_info_records(current_read):
                    # info_record[0] is the read name suffix
                    print(
                        read.name + info_record[0],
                        *info_record[1:],
                        self.RC_MAP[info.is_rc],
                        sep="\t",
                        file=self._file,
                    )
                current_read = match.trimmed(current_read)
        else:
            seq = read.sequence
            qualities = read.qualities if read.qualities is not None else ""
            print(read.name, -1, seq, qualities, sep="\t", file=self._file)

        return KEEP class SingleEndStep(ABC):
    @abstractmethod
    def __call__(self, read, info: ModificationInfo) -> bool:
        """
        Process a single read. Return True if the read has been consumed
        (and should thus not be passed on to subsequent steps).
        """ class PairedSingleEndStep(PairedEndStep):
    """
    Wrap a SingleEndStep as a PairedEndStep

    The wrapped step is called with the first read
    """

    def __init__(self, step: SingleEndStep):
        self._step = step

    def __repr__(self):
        return f"PairedSingleEndStep(step={self._step})"

    def __call__(self, read1, read2, info1, info2):
        _ = read2  # intentionally ignored
        _ = info2
        return self._step(read1, info1)
import io
import os
import sys
import copy
import logging
import multiprocessing.connection
import traceback
import dnaio
from typing import List, Optional, BinaryIO, TextIO, Any, Tuple, Dict, Sequence
from abc import ABC, abstractmethod
from multiprocessing import Process, Pipe, Queue
from pathlib import Path
from multiprocessing.connection import Connection
from .utils import Progress, FileOpener
from .modifiers import (
    SingleEndModifier,
    PairedEndModifier,
    PairedEndModifierWrapper,
    ModificationInfo,
)
from .report import Statistics
from .filters import (
    TooShort,
    TooLong,
    TooManyN,
    TooManyExpectedErrors,
    CasavaFiltered,
    DiscardTrimmed,
    DiscardUntrimmed,
    Predicate,
)
from .steps import (
    SingleEndSink,
    PairedEndSink,
    SingleEndFilter,
    PairedEndFilter,
    Demultiplexer,
    PairedDemultiplexer,
    CombinatorialDemultiplexer,
    RestFileWriter,
    WildcardFileWriter,
    InfoFileWriter,
    SingleEndStep,
    PairedSingleEndStep,
)

    def process_reads(
        self, progress: Progress = None
    ) -> Tuple[int, int, Optional[int]]:
        """Run the pipeline. Return statistics"""
        n = 0  # no. of processed reads
        total_bp = 0
        for read in self._reader:
            n += 1
            if n % 10000 == 0 and progress:
                progress.update(n)
            total_bp += len(read)
            info = ModificationInfo(read)
            for modifier in self._modifiers:
                read = modifier(read, info)
            for filter_ in self._steps:
                if filter_(read, info):
                    break
        return (n, total_bp, None)

    def _open_writer(
        self,
        file: BinaryIO,
        file2: Optional[BinaryIO] = None,
        force_fasta: Optional[bool] = None,
    ):
        assert file2 is None
        assert not isinstance(file, (str, bytes, Path))
        return self.file_opener.dnaio_open_raise_limit(
            file,
            mode="w",
            qualities=self.uses_qualities,
            fileformat="fasta" if force_fasta else None,
        )

    def _make_filter(self, writer, predicate1: Predicate, predicate2: Predicate):
        _ = predicate2
        return SingleEndFilter(writer, predicate1)

    def _make_untrimmed_filter(self, writer):
        return SingleEndFilter(writer, DiscardUntrimmed())

    def _final_filter(self, outfiles: OutputFiles):
        assert outfiles.out2 is None and outfiles.out is not None
        writer = self._open_writer(outfiles.out, force_fasta=outfiles.force_fasta)
        return SingleEndSink(writer)

    def _create_demultiplexer(self, outfiles: OutputFiles):
        writers = dict()  # type: Dict[Optional[str], Any]
        if outfiles.untrimmed is not None:
            writers[None] = self._open_writer(
                outfiles.untrimmed, force_fasta=outfiles.force_fasta
            )
        assert outfiles.demultiplex_out is not None
        for name, file in outfiles.demultiplex_out.items():
            writers[name] = self._open_writer(file, force_fasta=outfiles.force_fasta)
        return Demultiplexer(writers)

    def _wrap_single_end_step(self, step: SingleEndStep):
        return step

    @property
    def minimum_length(self):
        return self._minimum_length

    @minimum_length.setter
    def minimum_length(self, value):
        assert value is None or len(value) == 1
        self._minimum_length = value

    @property
    def maximum_length(self):
        return self._maximum_length

    @maximum_length.setter
    def maximum_length(self, value):
        assert value is None or len(value) == 1
        self._maximum_length = value


class PairedEndPipeline(Pipeline):
    """
    Processing pipeline for paired-end reads.
    """

    paired = True

    def __init__(self, file_opener: FileOpener, pair_filter_mode: str):
        super().__init__(file_opener)
        self._modifiers = []  # type: List[PairedEndModifier]
        self._pair_filter_mode = pair_filter_mode
        self._reader = None
        # Whether to ignore pair_filter mode for discard-untrimmed filter
        self.override_untrimmed_pair_filter = False

    def add(
        self,
        modifier1: Optional[SingleEndModifier],
        modifier2: Optional[SingleEndModifier],
    ) -> None:
        """
        Add a modifier for R1 and R2. One of them can be None, in which case the modifier
        will only be added for the respective read.
        """
        if modifier1 is None and modifier2 is None:
            raise ValueError("Not both modifiers can be None")
        self._modifiers.append(PairedEndModifierWrapper(modifier1, modifier2))

    def add_both(self, modifier: SingleEndModifier) -> None:
        """
        Add one modifier for both R1 and R2
        """
        assert modifier is not None
        self._modifiers.append(PairedEndModifierWrapper(modifier, copy.copy(modifier)))

    def add_paired_modifier(self, modifier: PairedEndModifier) -> None:
        """Add a Modifier (without wrapping it in a PairedEndModifierWrapper)"""
        self._modifiers.append(modifier)

    def process_reads(

        self, progress: Progress = None