comparison jbrowse2.py @ 73:3b2815efa5d9 draft

planemo upload for repository https://github.com/galaxyproject/tools-iuc/tree/master/tools/jbrowse2 commit b361309b05a861da9b64e1324157a8c32767e0bf

72:2bdb748df098

import json
import logging
import os
import re
import shutil
import struct
import subprocess
import tempfile
import urllib.request
import xml.etree.ElementTree as ET

    "}": "__cc__",
    "@": "__at__",
    "#": "__pd__",
    "": "__cn__",
}


class ColorScaling(object):

    COLOR_FUNCTION_TEMPLATE = """

class JbrowseConnector(object):
    def __init__(self, outdir, jbrowse2path):
        self.assemblies = []  # these require more than a few line diff.
        self.assmeta = {}
        self.giURL = GALAXY_INFRASTRUCTURE_URL
        self.outdir = outdir
        self.jbrowse2path = jbrowse2path
        os.makedirs(self.outdir, exist_ok=True)
        self.genome_names = []

            "displays": [
                style_data,
            ]
        }
        return wstyle

    def process_genomes(self, genomes):
        assembly = []
        assmeta = []
        useuri = False

            if genome_name not in genome_names:
                # pafs with shared references
                fapath = genome_node["path"]
                if not useuri:
                    fapath = os.path.realpath(fapath)
                assem = self.make_assembly(fapath, genome_name, useuri)
                assembly.append(assem)
                if len(genome_names) == 0:
                    this_genome["genome_name"] = genome_name  # first one for all tracks
                    genome_names.append(genome_name)
                    this_genome["genome_sequence_adapter"] = assem["sequence"][
                        "adapter"

                                ].strip()
                            else:
                                this_genome["genome_firstcontig"] = fl
                    else:
                        try:
                            fl = urllib.request.urlopen(fapath + ".fai").readline()
                        except Exception:
                            fl = None
                        if fl:  # is first row of the text fai so the first contig name
                            this_genome["genome_firstcontig"] = (
                                fl.decode("utf8").strip().split()[0]

        self.tracksToAdd[genome_names[0]] = []
        self.genome_names += genome_names
        return this_genome["genome_name"]

    def make_assembly(self, fapath, gname, useuri):
        if useuri:
            faname = fapath
            adapter = {
                "type": "BgzipFastaAdapter",
                "fastaLocation": {"uri": faname, "locationType": "UriLocation"},
                "faiLocation": {"uri": faname + ".fai", "locationType": "UriLocation"},
                "gziLocation": {"uri": faname + ".gzi", "locationType": "UriLocation"},
            }
        else:
            faname = gname + ".fa.gz"
            fadest = os.path.realpath(os.path.join(self.outdir, faname))
            cmd = "bgzip -i -c %s -I %s.gzi > %s && samtools faidx %s" % (
                fapath,

                },
                "gziLocation": {
                    "uri": faname + ".gzi",
                },
            }

        trackDict = {
            "name": gname,
            "sequence": {
                "type": "ReferenceSequenceTrack",
                "trackId": gname,

                    "type": "LinearGCContentDisplay",
                    "displayId": "%s-LinearGCContentDisplay" % gname,
                },
            ],
        }
        return trackDict

    def add_default_view(self):
        cmd = [
            "jbrowse",
            "set-default-session",

                        ["cp", os.path.realpath(cram_index), dest + ".crai"]
                    )
            else:
                cpath = os.path.realpath(dest) + ".crai"
                cmd = ["samtools", "index", "-c", "-o", cpath, os.path.realpath(dest)]
                logging.debug("executing cmd %s" % " ".join(cmd))
                self.subprocess_check_call(cmd)
        trackDict = {
            "type": "AlignmentsTrack",
            "trackId": tId,
            "name": trackData["name"],

                "https://"
            )

            if gname not in self.genome_names:
                # ignore if already there - eg for duplicates among pafs.
                asstrack = self.make_assembly(pgpaths[i], gname, useuri)
                self.genome_names.append(gname)
                self.tracksToAdd[gname] = []
                self.assemblies.append(asstrack)
        trackDict = {
            "type": "SyntenyTrack",

        for gnome in self.genome_names:
            tracks_data = []
            for track_conf in self.tracksToAdd[gnome]:
                tId = track_conf["trackId"]
                track_types[tId] = track_conf["type"]
                style_data = default_data["style"].get(tId, None)
                if not style_data:
                    logging.debug(
                        "### No style data in default data %s for %s"
                        % (default_data, tId)
                    )

        self.config_json.update(config_json)
        logging.debug("defaultSession=%s" % (pp))
        with open(self.config_json_file, "w") as config_file:
            json.dump(self.config_json, config_file, indent=2)

    def add_general_configuration(self, data):
        """
        Add some general configuration to the config.json file
        """

        config_json["configuration"].update(config_data)
        self.config_json.update(config_json)
        with open(config_path, "w") as config_file:
            json.dump(self.config_json, config_file, indent=2)

    def clone_jbrowse(self, realclone=True):
        """Clone a JBrowse directory into a destination directory. This also works in Biocontainer testing now
        Leave as True between version updates on temporary tools - requires manual conda trigger :(
        """
        dest = self.outdir
        if realclone:

        # be GET and not POST so it should redirect OK
        GALAXY_INFRASTRUCTURE_URL = "http://" + GALAXY_INFRASTRUCTURE_URL

    jc = JbrowseConnector(outdir=args.outdir, jbrowse2path=args.jbrowse2path)

    default_session_data = {
        "visibility": {
            "default_on": [],
            "default_off": [],
        },
        "style": {},
        "style_labels": {},
    }

    for ass in root.findall("assembly"):
        genomes = [
            {
                "path": x.attrib["path"],

                "meta": metadata_from_node(x.find("metadata")),
            }
            for x in ass.findall("metadata/genomes/genome")
        ]
        assref_name = jc.process_genomes(genomes)

        for track in ass.find("tracks"):
            track_conf = {}
            track_conf["trackfiles"] = []
            track_conf["assemblyNames"] = assref_name
            is_multi_bigwig = False

            if keys:
                for key in keys:
                    vis = track.attrib.get("visibility", "default_off")
                    if not vis:
                        vis = "default_off"
                    default_session_data["visibility"][vis].append(key)
                if track.find("options/style"):
                    default_session_data["style"][key] = {
                        item.tag: parse_style_conf(item)
                        for item in track.find("options/style")
                    }
                else:
                    default_session_data["style"][key] = {}
                    logging.debug("@@@@ no options/style found for %s" % (key))

                if track.find("options/style_labels"):
                    default_session_data["style_labels"][key] = {
                        item.tag: parse_style_conf(item)
                        for item in track.find("options/style_labels")
                    }
    default_session_data["defaultLocation"] = root.find(
        "metadata/general/defaultLocation"
    ).text
    default_session_data["session_name"] = root.find(
        "metadata/general/session_name"

        trackconf += jc.tracksToAdd[gnome]
    jc.config_json["tracks"] = trackconf
    assconf = jc.config_json.get("assemblies", [])
    assconf += jc.assemblies
    jc.config_json["assemblies"] = assconf
    logging.debug("&&&assemblies=%s, gnames=%s" % (assconf, jc.genome_names))
    jc.write_config()
    jc.add_default_session(default_session_data)
    # jc.text_index() not sure what broke here.

73:3b2815efa5d9

import json
import logging
import os
import re
import shutil
import ssl
import struct
import subprocess
import tempfile
import urllib.request
import xml.etree.ElementTree as ET

    "}": "__cc__",
    "@": "__at__",
    "#": "__pd__",
    "": "__cn__",
}


INDEX_TEMPLATE = """<!doctype html>
<html lang="en" style="height:100%">
<head>
<meta charset="utf-8"/>
<link rel="shortcut icon" href="./favicon.ico"/>
<meta name="viewport" content="width=device-width,initial-scale=1"/>
<meta name="theme-color" content="#000000"/>
<meta name="description" content="A fast and flexible genome browser"/>
<link rel="manifest" href="./manifest.json"/>
<title>JBrowse</title>
</script>
</head>
<body style="overscroll-behavior:none; height:100%; margin: 0;">
<iframe
  id="jbframe"
  title="JBrowse2"
  frameborder="0"
  width="100%"
  height="100%"
  src='index_noview.html?config=config.json__SESSION_SPEC__'>
</iframe>
</body>
</html>
"""


class ColorScaling(object):

    COLOR_FUNCTION_TEMPLATE = """

class JbrowseConnector(object):
    def __init__(self, outdir, jbrowse2path):
        self.assemblies = []  # these require more than a few line diff.
        self.assmeta = {}
        self.ass_first_contigs = (
            []
        )  # for default session - these are read as first line of the assembly .fai
        self.giURL = GALAXY_INFRASTRUCTURE_URL
        self.outdir = outdir
        self.jbrowse2path = jbrowse2path
        os.makedirs(self.outdir, exist_ok=True)
        self.genome_names = []

            "displays": [
                style_data,
            ]
        }
        return wstyle

    def urllib_get_2018():
        # Using a protected member like this is not any more fragile
        # than extending the class and using it. I would use it.
        url = "https://localhost:6667/my-endpoint"
        ssl._create_default_https_context = ssl._create_unverified_context
        with urllib.request.urlopen(url=url) as f:
            print(f.read().decode("utf-8"))

    def urllib_get_2022():
        # Finally! Able to use the publice API. Happy happy!
        url = "https://localhost:6667/my-endpoint"
        scontext = ssl.SSLContext(ssl.PROTOCOL_TLS)
        scontext.verify_mode = ssl.VerifyMode.CERT_NONE
        with urllib.request.urlopen(url=url, context=scontext) as f:
            print(f.read().decode("utf-8"))

    def process_genomes(self, genomes):
        assembly = []
        assmeta = []
        useuri = False

            if genome_name not in genome_names:
                # pafs with shared references
                fapath = genome_node["path"]
                if not useuri:
                    fapath = os.path.realpath(fapath)
                assem, first_contig = self.make_assembly(fapath, genome_name, useuri)
                assembly.append(assem)
                self.ass_first_contigs.append(first_contig)
                if len(genome_names) == 0:
                    this_genome["genome_name"] = genome_name  # first one for all tracks
                    genome_names.append(genome_name)
                    this_genome["genome_sequence_adapter"] = assem["sequence"][
                        "adapter"

                                ].strip()
                            else:
                                this_genome["genome_firstcontig"] = fl
                    else:
                        try:
                            scontext = ssl.SSLContext(ssl.PROTOCOL_TLS)
                            scontext.verify_mode = ssl.VerifyMode.CERT_NONE
                            with urllib.request.urlopen(
                                url=fapath + ".fai", context=scontext
                            ) as f:
                                fl = f.readline()
                        except Exception:
                            fl = None
                        if fl:  # is first row of the text fai so the first contig name
                            this_genome["genome_firstcontig"] = (
                                fl.decode("utf8").strip().split()[0]

        self.tracksToAdd[genome_names[0]] = []
        self.genome_names += genome_names
        return this_genome["genome_name"]

    def make_assembly(self, fapath, gname, useuri):
        """added code to grab the first contig name and length for broken default session from Anthony and Helena's code
        that poor Bjoern is trying to figure out.
        """
        if useuri:
            faname = fapath
            adapter = {
                "type": "BgzipFastaAdapter",
                "fastaLocation": {"uri": faname, "locationType": "UriLocation"},
                "faiLocation": {"uri": faname + ".fai", "locationType": "UriLocation"},
                "gziLocation": {"uri": faname + ".gzi", "locationType": "UriLocation"},
            }
            scontext = ssl.SSLContext(ssl.PROTOCOL_TLS)
            scontext.verify_mode = ssl.VerifyMode.CERT_NONE
            with urllib.request.urlopen(url=faname + ".fai", context=scontext) as f:
                fl = f.readline()
            contig = fl.decode("utf8").strip()
            # Merlin  172788  8       60      61
        else:
            faname = gname + ".fa.gz"
            fadest = os.path.realpath(os.path.join(self.outdir, faname))
            cmd = "bgzip -i -c %s -I %s.gzi > %s && samtools faidx %s" % (
                fapath,

                },
                "gziLocation": {
                    "uri": faname + ".gzi",
                },
            }
            contig = open(fadest + ".fai", "r").readline().strip()
        first_contig = contig.split()[:2]
        first_contig.insert(0, gname)
        trackDict = {
            "name": gname,
            "sequence": {
                "type": "ReferenceSequenceTrack",
                "trackId": gname,

                    "type": "LinearGCContentDisplay",
                    "displayId": "%s-LinearGCContentDisplay" % gname,
                },
            ],
        }
        return (trackDict, first_contig)

    def add_default_view(self):
        cmd = [
            "jbrowse",
            "set-default-session",

                        ["cp", os.path.realpath(cram_index), dest + ".crai"]
                    )
            else:
                cpath = os.path.realpath(dest) + ".crai"
                cmd = ["samtools", "index", "-c", "-o", cpath, os.path.realpath(dest)]
                self.subprocess_check_call(cmd)
        trackDict = {
            "type": "AlignmentsTrack",
            "trackId": tId,
            "name": trackData["name"],

                "https://"
            )

            if gname not in self.genome_names:
                # ignore if already there - eg for duplicates among pafs.
                asstrack, first_contig = self.make_assembly(pgpaths[i], gname, useuri)
                self.genome_names.append(gname)
                self.tracksToAdd[gname] = []
                self.assemblies.append(asstrack)
        trackDict = {
            "type": "SyntenyTrack",

        for gnome in self.genome_names:
            tracks_data = []
            for track_conf in self.tracksToAdd[gnome]:
                tId = track_conf["trackId"]
                track_types[tId] = track_conf["type"]
                style_data = default_data[gnome]["style"].get(tId, None)
                if not style_data:
                    logging.debug(
                        "### No style data in default data %s for %s"
                        % (default_data, tId)
                    )

        self.config_json.update(config_json)
        logging.debug("defaultSession=%s" % (pp))
        with open(self.config_json_file, "w") as config_file:
            json.dump(self.config_json, config_file, indent=2)

    def add_defsess_to_index(self, data):
        """
        Broken in Anthony's PR because only ever dealt with the first assembly.

        Add some default session settings: set some assemblies/tracks on/off

        This allows to select a default view:
        - jb type (Linear, Circular, etc)
        - default location on an assembly
        - default tracks
        - ...

        Different methods to do that were tested/discussed:
        - using a defaultSession item in config.json: this proved to be difficult:
          forced to write a full session block, including hard-coded/hard-to-guess items,
          no good way to let Jbrowse2 display a scaffold without knowing its size
        - using JBrowse2 as an embedded React component in a tool-generated html file:
          it works but it requires generating js code to actually do what we want = chosing default view, assembly, tracks, ...
        - writing a session-spec inside the config.json file: this is not yet supported as of 2.10.2 (see PR 4148 below)
          a session-spec is a kind of simplified defaultSession where you don't need to specify every aspect of the session
        - passing a session-spec through URL params by embedding the JBrowse2 index.html inside an iframe
          we selected this option

        Xrefs to understand the choices:
        https://github.com/GMOD/jbrowse-components/issues/2708
        https://github.com/GMOD/jbrowse-components/discussions/3568
        https://github.com/GMOD/jbrowse-components/pull/4148
        """
        new_index = "Nothing written"
        session_spec = {"views": []}
        logging.debug("def data=%s" % data)
        for first_contig in self.ass_first_contigs:
            gnome, refName, end = first_contig
            start = 0
            if False or data.get("defaultLocation", ""):
                loc_match = re.search(
                    r"^([^:]+):([\d,]*)\.*([\d,]*)$", data["defaultLocation"]
                )
                # loc_match = re.search(r"^(\w+):(\d+)\.+(\d+)$", data["defaultLocation"])
                if loc_match:
                    refName = loc_match.group(1)
                    start = int(loc_match.group(2))
                    end = int(loc_match.group(3))
            else:
                if refName:
                    view = {
                        "assembly": gnome,
                        "loc": "{}:{}-{}".format(refName, start, end),
                        "type": "LinearGenomeView",
                        "tracks": data[gnome]["tracks"],
                    }
                    session_spec["views"].append(view)
        sess = json.dumps(session_spec, sort_keys=True, indent=2)
        new_index = INDEX_TEMPLATE.replace(
            "__SESSION_SPEC__", "&session=spec-{}".format(sess)
        )

        os.rename(
            os.path.join(self.outdir, "index.html"),
            os.path.join(self.outdir, "index_noview.html"),
        )

        with open(os.path.join(self.outdir, "index.html"), "w") as nind:
            nind.write(new_index)
        logging.debug(
            "#### add_defsession gnome=%s refname=%s\nsession_spec=%s\nnew_index=%s"
            % (gnome, refName, sess, new_index)
        )

    def add_general_configuration(self, data):
        """
        Add some general configuration to the config.json file
        """

        config_json["configuration"].update(config_data)
        self.config_json.update(config_json)
        with open(config_path, "w") as config_file:
            json.dump(self.config_json, config_file, indent=2)

    def clone_jbrowse(self, realclone=False):
        """Clone a JBrowse directory into a destination directory. This also works in Biocontainer testing now
        Leave as True between version updates on temporary tools - requires manual conda trigger :(
        """
        dest = self.outdir
        if realclone:

        # be GET and not POST so it should redirect OK
        GALAXY_INFRASTRUCTURE_URL = "http://" + GALAXY_INFRASTRUCTURE_URL

    jc = JbrowseConnector(outdir=args.outdir, jbrowse2path=args.jbrowse2path)

    default_session_data = {}

    for ass in root.findall("assembly"):
        genomes = [
            {
                "path": x.attrib["path"],

                "meta": metadata_from_node(x.find("metadata")),
            }
            for x in ass.findall("metadata/genomes/genome")
        ]
        assref_name = jc.process_genomes(genomes)
        if not default_session_data.get(assref_name, None):
            default_session_data[assref_name] = {
                "tracks": [],
                "style": {},
                "style_labels": {},
                "visibility": {
                    "default_on": [],
                    "default_off": [],
                },
            }
        for track in ass.find("tracks"):
            track_conf = {}
            track_conf["trackfiles"] = []
            track_conf["assemblyNames"] = assref_name
            is_multi_bigwig = False

            if keys:
                for key in keys:
                    vis = track.attrib.get("visibility", "default_off")
                    if not vis:
                        vis = "default_off"
                    default_session_data[assref_name]["visibility"][vis].append(key)
                if track.find("options/style"):
                    default_session_data[assref_name]["style"][key] = {
                        item.tag: parse_style_conf(item)
                        for item in track.find("options/style")
                    }
                else:
                    default_session_data[assref_name]["style"][key] = {}
                    logging.debug("no options/style found for %s" % (key))

                if track.find("options/style_labels"):
                    default_session_data[assref_name]["style_labels"][key] = {
                        item.tag: parse_style_conf(item)
                        for item in track.find("options/style_labels")
                    }
                default_session_data[assref_name]["tracks"].append(key)
    default_session_data["defaultLocation"] = root.find(
        "metadata/general/defaultLocation"
    ).text
    default_session_data["session_name"] = root.find(
        "metadata/general/session_name"

        trackconf += jc.tracksToAdd[gnome]
    jc.config_json["tracks"] = trackconf
    assconf = jc.config_json.get("assemblies", [])
    assconf += jc.assemblies
    jc.config_json["assemblies"] = assconf
    logging.debug("assemblies=%s, gnames=%s" % (assconf, jc.genome_names))
    jc.write_config()
    jc.add_default_session(default_session_data)
    jc.add_defsess_to_index(default_session_data)
    # jc.text_index() not sure what broke here.