Source code for galaxy.datatypes.converters.interval_to_fli

"""
Creates a feature location index (FLI) for a given BED/GFF file.
FLI index has the form::

    [line_length]
    <symbol1_in_lowercase><tab><symbol1><tab><location>
    <symbol2_in_lowercase><tab><symbol2><tab><location>
    ...

where location is formatted as:

    contig:start-end

and symbols are sorted in lexigraphical order.
"""

import optparse

from bx.tabular.io import (
    Comment,
    Header,
)

from galaxy.datatypes.util.gff_util import (
    convert_gff_coords_to_bed,
    GFFReaderWrapper,
    read_unordered_gtf,
)


[docs]def main():
    # Process arguments.
    parser = optparse.OptionParser()
    parser.add_option("-F", "--format", dest="input_format")
    (options, args) = parser.parse_args()
    in_fname, out_fname = args
    input_format = options.input_format.lower()

    # Create dict of name-location pairings.
    name_loc_dict = {}
    if input_format in ["gff", "gtf"]:
        # GTF/GFF format

        # Create reader.
        if input_format == "gff":
            in_reader = GFFReaderWrapper(open(in_fname))
        else:  # input_format == 'gtf'
            in_reader = read_unordered_gtf(open(in_fname))

        for feature in in_reader:
            if isinstance(feature, (Header, Comment)):
                continue

            for name in feature.attributes:
                val = feature.attributes[name]
                try:
                    float(val)
                    continue
                except ValueError:
                    convert_gff_coords_to_bed(feature)
                    # Value is not a number, so it can be indexed.
                    if val not in name_loc_dict:
                        # Value is not in dictionary.
                        name_loc_dict[val] = {"contig": feature.chrom, "start": feature.start, "end": feature.end}
                    else:
                        # Value already in dictionary, so update dictionary.
                        loc = name_loc_dict[val]
                        if feature.start < loc["start"]:
                            loc["start"] = feature.start
                        if feature.end > loc["end"]:
                            loc["end"] = feature.end
    elif input_format == "bed":
        # BED format.
        for line in open(in_fname):
            # Ignore track lines.
            if line.startswith("track"):
                continue

            fields = line.split()

            # Ignore lines with no feature name.
            if len(fields) < 4:
                continue

            # Process line
            name_loc_dict[fields[3]] = {"contig": fields[0], "start": int(fields[1]), "end": int(fields[2])}

    # Create sorted list of entries.
    max_len = 0
    entries = []
    for name in sorted(name_loc_dict.keys()):
        loc = name_loc_dict[name]
        entry = "{}\t{}\t{}".format(name.lower(), name, "%s:%i-%i" % (loc["contig"], loc["start"], loc["end"]))
        if len(entry) > max_len:
            max_len = len(entry)
        entries.append(entry)

    # Write padded entries.
    with open(out_fname, "w") as out:
        out.write(f"{str(max_len + 1).ljust(max_len)}\n")
        for entry in entries:
            out.write(f"{entry.ljust(max_len)}\n")


if __name__ == "__main__":
    main()