comparison diagonal_partition.py @ 4:36cafb694dd2 draft

planemo upload for repository https://github.com/richard-burhans/galaxytools/tree/main/tools/segalign commit b8aa943b38b865defab8a27e4404bb8a2131f919

3:6f46cebc9ed8

diagonal_partition <max-segments> <lastz-command>
"""

import os
import sys


def chunks(lst, n):
    """Yield successive n-sized chunks from list."""
    for i in range(0, len(lst), n):
        yield lst[i: i + n]

    params = sys.argv[2:]

    # Parsing command output from SegAlign
    segment_key = "--segments="
    segment_index = None
    input_file = None

    for index, value in enumerate(params):
        if value[: len(segment_key)] == segment_key:
            segment_index = index
            input_file = value[len(segment_key):]
            break
    if segment_index is None:
        print(f"Error: could not segment key {segment_key} in parameters {params}")
        exit(0)

    if not os.path.isfile(input_file):
        print(f"Error: File {input_file} does not exist")
        exit(0)

    if (
        chunk_size == 0 or sum(1 for _ in open(input_file)) <= chunk_size

        print(" ".join(params), flush=True)
        exit(0)

    # Find rest of relevant parameters
    output_key = "--output="
    output_index = None
    output_alignment_file = None
    output_alignment_file_base = None
    output_format = None

    strand_key = "--strand="
    strand_index = None
    for index, value in enumerate(params):

        exit(0)

    err_index = -1  # error file is at very end
    err_name_base = params[-1].split(".err", 1)[0]

    data = {}  # dict of list of tuple (x, y, str)

    direction = None
    if "plus" in params[strand_index]:
        direction = "f"
    elif "minus" in params[strand_index]:

            with open(fname, "w") as f:
                f.writelines(chunk)
            # update segment file in command
            params[segment_index] = segment_key + fname
            # update output file in command
            params[output_index] = (
                output_key
                + output_alignment_file_base
                + name_addition
                + "."
                + output_format
            )
            # update error file in command
            params[-1] = err_name_base + name_addition + ".err"
            print(" ".join(params), flush=True)

    # writing unsorted skipped pairs
    if len(skip_pairs) > 0:
        skip_pairs_with_len = sorted(
            [(len(data[p]), p) for p in skip_pairs]
        )  # list of tuples of (pair length, pair)
        aggregated_skip_pairs = []  # list of list of pair names
        current_count = 0
        aggregated_skip_pairs.append([])
        for count, pair in skip_pairs_with_len:
            if current_count + count <= chunk_size:
                current_count += count

                    chunk = list(zip(*data[pair]))[2]
                    f.writelines(chunk)
            # update segment file in command
            params[segment_index] = segment_key + fname
            # update output file in command
            params[output_index] = (
                output_key
                + output_alignment_file_base
                + name_addition
                + "."
                + output_format
            )
            # update error file in command
            params[-1] = err_name_base + name_addition + ".err"
            print(" ".join(params), flush=True)

    if DELETE_AFTER_CHUNKING:

4:36cafb694dd2

diagonal_partition <max-segments> <lastz-command>
"""

import os
import sys
import typing

T = typing.TypeVar("T", bound="_Sliceable")


class _Sliceable(typing.Protocol):
    def __len__(self) -> int:
        ...

    def __getitem__(self: T, i: slice) -> T:
        ...


def chunks(lst: T, n: int) -> typing.Iterator[T]:
    """Yield successive n-sized chunks from list."""
    for i in range(0, len(lst), n):
        yield lst[i: i + n]

    params = sys.argv[2:]

    # Parsing command output from SegAlign
    segment_key = "--segments="
    segment_index = None
    input_file: typing.Optional[str] = None

    for index, value in enumerate(params):
        if value[: len(segment_key)] == segment_key:
            segment_index = index
            input_file = value[len(segment_key):]
            break
    if segment_index is None:
        print(f"Error: could not segment key {segment_key} in parameters {params}")
        exit(0)

    if input_file is None or not os.path.isfile(input_file):
        print(f"Error: File {input_file} does not exist")
        exit(0)

    if (
        chunk_size == 0 or sum(1 for _ in open(input_file)) <= chunk_size

        print(" ".join(params), flush=True)
        exit(0)

    # Find rest of relevant parameters
    output_key = "--output="
    output_index: typing.Optional[int] = None
    output_alignment_file = None
    output_alignment_file_base: typing.Optional[str] = None
    output_format = None

    strand_key = "--strand="
    strand_index = None
    for index, value in enumerate(params):

        exit(0)

    err_index = -1  # error file is at very end
    err_name_base = params[-1].split(".err", 1)[0]

    data: typing.Dict[
        typing.Tuple[str, str], typing.List[typing.Tuple[int, int, str]]
    ] = {}  # dict of list of tuple (x, y, str)

    direction = None
    if "plus" in params[strand_index]:
        direction = "f"
    elif "minus" in params[strand_index]:

            with open(fname, "w") as f:
                f.writelines(chunk)
            # update segment file in command
            params[segment_index] = segment_key + fname
            # update output file in command
            if output_index is not None:
                params[output_index] = (
                    f"{output_key}{output_alignment_file_base}{name_addition}.{output_format}"
                )
            # update error file in command
            params[-1] = err_name_base + name_addition + ".err"
            print(" ".join(params), flush=True)

    # writing unsorted skipped pairs
    if len(skip_pairs) > 0:
        skip_pairs_with_len = sorted(
            [(len(data[p]), p) for p in skip_pairs]
        )  # list of tuples of (pair length, pair)
        aggregated_skip_pairs: typing.List[typing.List[typing.Any]] = (
            []
        )  # list of list of pair names
        current_count = 0
        aggregated_skip_pairs.append([])
        for count, pair in skip_pairs_with_len:
            if current_count + count <= chunk_size:
                current_count += count

                    chunk = list(zip(*data[pair]))[2]
                    f.writelines(chunk)
            # update segment file in command
            params[segment_index] = segment_key + fname
            # update output file in command
            if output_index is not None:
                params[output_index] = (
                    f"{output_key}{output_alignment_file_base}{name_addition}.{output_format}"
                )
            # update error file in command
            params[-1] = err_name_base + name_addition + ".err"
            print(" ".join(params), flush=True)

    if DELETE_AFTER_CHUNKING: