Source code for higlass.server

from functools import partial
from io import StringIO
import multiprocess as mp
import cytoolz as toolz
import os.path as op
import platform
import logging
import logging.handlers
import socket
import tempfile
import json
import time
import os

from flask import Flask
from flask import request, jsonify

# from flask_restful import reqparse, abort, Api, Resource
from flask_cors import CORS

import requests
import slugid
import sh

__all__ = ["Server"]

OS_NAME = platform.system()
OS_TEMPDIR = tempfile.gettempdir()

# Disable annoying logs from werkzeug server
log = logging.getLogger("werkzeug")
log.disabled = True

# The following line is also needed to turn off all debug logs
os.environ["WERKZEUG_RUN_MAIN"] = "true"

def create_app(name, tilesets, fuse=None):
    app = Flask(name)
    app.logger.disabled = True

    remote_tilesets = {}

    def hello():
        return "Hello World!"

    @app.route("/api/v1/register_url/", methods=["POST"])
    def register_url():
        from higlass.tilesets import by_filetype

        js = request.json
        if js["filetype"] not in by_filetype:
            return (
                jsonify({"error": "Unknown filetype: {}".format(js["filetype"])}),

        if fuse is None:
            return jsonify({"error": "httpfs is not available."})

        key = (js["fileUrl"], js["filetype"])
        if key in remote_tilesets:
            ts = remote_tilesets[key]
            mounted_url = fuse.get_filepath(js["fileUrl"])
            factory = by_filetype[js["filetype"]]
            ts = factory(mounted_url)
            remote_tilesets[key] = ts

        return jsonify({"uid": ts.uuid})

    @app.route("/api/v1/available-chrom-sizes/", methods=["GET"])
    def available_chrom_sizes():
        Get the list of available chromosome size lists. No query parameters.

        results = []
        for ts in tilesets:
            if ts.datatype == "chromsizes":
        return jsonify({"count": len(results), "results": results})

    @app.route("/api/v1/chrom-sizes/", methods=["GET"])
    def chrom_sizes():
        Coordinate system resource.

        Query Parameters
        id : string
            Tileset UUID
        type : { tsv | json }
            Response format. Default is tsv.
        cum : boolean
            Return cumulative lengths. Default is false.

        uuid = request.args.get("id", None)
        res_type = request.args.get("type", "tsv")
        incl_cum = request.args.get("cum", False)

        # filter for tileset
        ts = next((ts for ts in _list_tilesets() if ts.uuid == uuid), None)
        if ts is None:
            return jsonify({"error": "Not found"}), 404
        if not hasattr(ts, "chromsizes"):
            return jsonify({"error": "Tileset does not have chrom sizes."})

        # list of tuples (chrom, size)
        data = ts.chromsizes
        if incl_cum:
            data, _data, cum = [], data, 0
            for chrom, size in _data:
                cum += size
                data.append((chrom, size, cum))

        if res_type == "json":
            if incl_cum:
                j = {
                    ts.uuid: {
                        chrom: {"size": size, "offset": offset}
                        for chrom, size, offset in data
                j = {ts.uuid: {chrom: {"size": size} for chrom, size in data}}
            return jsonify(j)
        elif res_type == "tsv":
            if incl_cum:
                return "\n".join(
                    "{}\t{}\t{}".format(chrom, size, offset)
                    for chrom, size, offset in data
                return "\n".join("{}\t{}".format(chrom, size) for chrom, size in data)
            return jsonify({"error": "Unknown response type"}), 500

    @app.route("/api/v1/uids_by_filename/", methods=["GET"])
    def uids_by_filename():
        return jsonify(
                "count": len(tilesets),
                "results": {i: tilesets[i] for i in range(len(tilesets))},

    def _list_tilesets():
        return tilesets + list(remote_tilesets.values())

    @app.route("/api/v1/tilesets/", methods=["GET"])
    def list_tilesets():
        tsets = _list_tilesets()
        return jsonify(
                "count": len(tsets),
                "next": None,
                "previous": None,
                "results": [ts.meta for ts in tsets],

    @app.route("/api/v1/tileset_info/", methods=["GET"])
    def tileset_info():
        uuids = request.args.getlist("d")

        info = {}
        for uuid in uuids:
            ts = next((ts for ts in _list_tilesets() if ts.uuid == uuid), None)

            if ts is not None:
                info[uuid] = ts.tileset_info()
                info[uuid] = {"error": "No such tileset with uid: {}".format(uuid)}

        return jsonify(info)

    @app.route("/api/v1/tiles/", methods=["GET"])
    def tiles():
        tids_requested = set(request.args.getlist("d"))

        if not tids_requested:
            return jsonify({"error": "No tiles requested"}), 400

        extract_uuid = lambda tid: tid.split(".")[0]
        uuids_to_tids = toolz.groupby(extract_uuid, tids_requested)

        tiles = []
        for uuid, tids in uuids_to_tids.items():
            ts = next((ts for ts in _list_tilesets() if ts.uuid == uuid), None)
        data = {tid: tval for tid, tval in tiles}
        return jsonify(data)

    return app

class ServerError(Exception):

def get_open_port():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(("", 0))
    port = s.getsockname()[1]
    return port

class FuseProcess:
    def __init__(self, tmp_dir):
        self.tmp_dir = tmp_dir
        self.http_directory = op.join(tmp_dir, "http")
        self.https_directory = op.join(tmp_dir, "https")
        self.diskcache_directory = op.join(tmp_dir, "dc")

    def setup(self):
        Set up filesystem in user space for http and https
        so that we can retrieve tiles from remote sources.

        tmp_dir: string
            The temporary directory where to create the
            http and https directories
        from simple_httpfs import HttpFs

        if not op.exists(self.http_directory):
        if not op.exists(self.https_directory):
        if not op.exists(self.diskcache_directory):


        disk_cache_size = 2 ** 25
        lru_capacity = 400

        def start_fuse(directory, protocol):
                # this import can cause problems on systems that don't have libfuse
                # installed so let's only try it if absolutely necessary
                from fuse import FUSE

                # This is a bit confusing. I think `fuse` (lowercase) is used
                # above in get_filepath() line 50 and 52. If that's not the
                # case than this assignment is useless and get_filepath() is
                # broken
                fuse = FUSE(
                    # allow_other=True
            except RuntimeError as e:
                if str(e) != "1":
                    raise e

        proc1 = mp.Process(target=start_fuse, args=[self.http_directory, "http"])

        proc2 = mp.Process(target=start_fuse, args=[self.https_directory, "https"])

    def teardown(self):
            if OS_NAME == "Darwin":
                sh.fusermount("-uz", self.http_directory)
        except Exception:

            if OS_NAME == "Darwin":
                sh.fusermount("-uz", self.https_directory)
        except Exception:

    def get_filepath(self, url):
        Get the httpfs mount filepath from a url

        if url[:7] == "http://":
            return self.http_directory + url[6:] + ".."
        elif url[:8] == "https://":
            return self.https_directory + url[7:] + ".."
            raise ValueError("Unsupported URL protocol")

[docs]class Server: """ A lightweight HiGlass server. Parameters ---------- tilesets : list A list of tilesets to serve (see higlass.tilesets) host : str, optional The host this server is running on. Usually just localhost. port : int, optional The port that this server will run on. name : str, optional A name for the Flask app being served. If not provided, a unique name will be generated. The app's logger inherits this name. fuse : bool, optional Whether to mount http(s) resources using FUSE. tmp_dir : string, optional A temporary directory for FUSE to mount the http(s) files and for caching. log_level: logging.* What level to log at log_file: str, optional Where to write diagnostic log files. Default is to use a StringIO stream in memory. """ # Keep track of the server processes that have been started. # So that when someone says 'start', the old ones are terminated processes = {} def __init__( self, tilesets, host="localhost", port=None, name=None, fuse=True, tmp_dir=OS_TEMPDIR, log_level=logging.INFO, log_file=None, ): = name or __name__.split(".")[0] + "-" + slugid.nice()[:8] self.tilesets = tilesets = host self.port = port if fuse: self.fuse_process = FuseProcess(op.join(tmp_dir, "higlass-python")) self.fuse_process.setup() else: self.fuse_process = None = create_app(, self.tilesets, fuse=self.fuse_process) if log_file: self.log = None handler = logging.handlers.RotatingFileHandler( log_file, maxBytes=100000, backupCount=1 ) else: self.log = StringIO() handler = logging.StreamHandler(self.log) handler.setLevel(log_level)
[docs] def start(self, debug=False, **kwargs): """ Start a lightweight higlass server. Parameters ---------- debug: bool Run the server in debug mode. Default is False. kwargs : Additional options to pass to """ for puid in list(self.processes.keys()): self.processes[puid].terminate() del self.processes[puid] if self.port is None: self.port = get_open_port() # we're going to assign a uuid to each server process so that if # anything goes wrong, the variable referencing the process doesn't get # lost uuid = slugid.nice() target = partial(, debug=debug,, port=self.port, threaded=True, use_reloader=False, **kwargs ) self.processes[uuid] = mp.Process(target=target) self.processes[uuid].start() self.connected = False while not self.connected: try: url = "http://{}:{}/api/v1".format(, self.port) r = requests.head(url) if r.ok: self.connected = True except requests.ConnectionError: time.sleep(0.2)
[docs] def stop(self): """ Stop this server so that the calling process can exit """ if self.fuse_process is not None: self.fuse_process.teardown() for uuid in self.processes: self.processes[uuid].terminate()
[docs] def tileset_info(self, uid): """ Return the tileset info for the given tileset """ url = "http://{host}:{port}/api/v1/tileset_info/?d={uid}".format(, port=self.port, uid=uid ) req = requests.get(url) if req.status_code != 200: raise ServerError("Error fetching tileset_info:", req.content) content = json.loads(req.content) return content[uid]
[docs] def tiles(self, uid, z, x, y=None): """ Return tiles from the specified dataset (uid) at the given position (z,x,[u]) """ tile_id = "{uid}.{z}.{x}".format(uid=uid, z=z, x=x) if y is not None: tile_id += ".{y}".format(y=y) url = "http://{host}:{port}/api/v1/tiles/?d={tile_id}".format(, port=self.port, tile_id=tile_id ) req = requests.get(url) if req.status_code != 200: raise ServerError("Error fetching tile:", req.content) content = json.loads(req.content) return content[tile_id]
[docs] def chromsizes(self, uid): """ Return the chromosome sizes from the given filename """ url = "http://{host}:{port}/api/v1/chrom-sizes/?id={uid}".format(, port=self.port, uid=uid ) req = requests.get(url) if req.status_code != 200: raise ServerError("Error fetching chromsizes:", req.content) return req.content
@property def api_address(self): return "http://{host}:{port}/api/v1".format(, port=self.port)