refactor visualization methods

src/framed/experimental/metanetx.py

@@ -26,6 +26,7 @@ def _get_reac_xref(self):
             filename = self.path_to_files + '/reac_xref.tsv'
             self.reac_xref = pd.read_csv(filename, sep='\t', header=None, comment='#',
                                          names=['external_id', 'mnx_id'])
+
             if self.version >= 3.0:
                 self.reac_xref = self.reac_xref[self.reac_xref['external_id'].str.contains(':')]
 

src/framed/experimental/visualization.py → src/framed/experimental/visualization/KEGG.py

@@ -1,48 +1,3 @@
-from __future__ import division
-from future import standard_library
-standard_library.install_aliases()
-
-
-
-def build_escher_map(fluxes, map_name, abstol=1e-6):
-    """ Build escher map for a given flux distribution
-
-    Args:
-        fluxes (dict): flux distribution
-        map_name (str): name of **escher** map (for a list of maps see *list_escher_maps*)
-        abstol (float): tolerance to remove residual fluxes for cleaner visualization (default: 1e-6)
-
-    Returns:
-        escher.plots.Builder: escher map object
-    """
-
-    import escher
-
-    data = {r_id[2:]: abs(val) if abs(val) > abstol else 0.0 for r_id, val in fluxes.items()}
-
-    colors = [{'type': 'min', 'color': '#f0f0f0', 'size': 8},
-              {'type': 'mean', 'color': '#abb7ff', 'size': 20},
-              {'type': 'max', 'color': '#0f16fb', 'size': 40}]
-
-    emap = escher.Builder(map_name=map_name, reaction_data=data,
-                          reaction_scale=colors,
-                          reaction_no_data_size=8,
-                          reaction_no_data_color='#ffddda')
-
-    return emap
-
-
-def list_escher_maps():
-    """ List of maps available in **escher**
-
-    Returns:
-        list: map names
-    """
-    import escher
-
-    maps = escher.list_available_maps()
-    return [entry['map_name'] for entry in maps]
-
 
 def highligh_enzymes_in_KEGG(pathway_id, enzymes, ax=None, color="#FF1414"):
     """ Highlight enzymes in KEGG map
@@ -59,7 +14,6 @@ def highligh_enzymes_in_KEGG(pathway_id, enzymes, ax=None, color="#FF1414"):
     from io import StringIO
     import requests
     import xml.etree.ElementTree as xml
-    import PIL.Image
     import PIL.ImageDraw
     import re
     import pandas as pd

src/framed/experimental/visualization/escher.py

@@ -0,0 +1,39 @@
+def build_escher_map(fluxes, map_name, abstol=1e-6):
+    """ Build escher map for a given flux distribution
+
+    Args:
+        fluxes (dict): flux distribution
+        map_name (str): name of **escher** map (for a list of maps see *list_escher_maps*)
+        abstol (float): tolerance to remove residual fluxes for cleaner visualization (default: 1e-6)
+
+    Returns:
+        escher.plots.Builder: escher map object
+    """
+
+    import escher
+
+    data = {r_id[2:]: abs(val) if abs(val) > abstol else 0.0 for r_id, val in fluxes.items()}
+
+    colors = [{'type': 'min', 'color': '#f0f0f0', 'size': 8},
+              {'type': 'mean', 'color': '#abb7ff', 'size': 20},
+              {'type': 'max', 'color': '#0f16fb', 'size': 40}]
+
+    emap = escher.Builder(map_name=map_name, reaction_data=data,
+                          reaction_scale=colors,
+                          reaction_no_data_size=8,
+                          reaction_no_data_color='#ffddda')
+
+    return emap
+
+
+def list_escher_maps():
+    """ List of maps available in **escher**
+
+    Returns:
+        list: map names
+    """
+    import escher
+
+    maps = escher.list_available_maps()
+    return [entry['map_name'] for entry in maps]
+

src/framed/experimental/visualization/iPATH.py

@@ -0,0 +1,103 @@
+from __future__ import division
+import requests
+from math import ceil
+import numpy as np
+
+from framed.experimental.metanetx import MetaNetX
+
+
+def iPATH_display(data, output='svg', filename=None, default_opacity=0.1, default_width=2, default_radius=5, **kwargs):
+
+    url = 'https://pathways.embl.de/mapping.cgi'
+
+    elements = edge_mapper(data)
+
+    parameters = {
+        'map': 'metabolic',
+        'export_type': 'svg',
+        'selection': '\n'.join(elements),
+        'default_opacity': default_opacity,
+        'default_width': default_width,
+        'default_radius': default_radius
+    }
+
+    parameters.update(kwargs)
+
+    r = requests.post(url, data=parameters)
+
+    if r.headers['Content-Type'] != 'image/svg+xml':
+        print(r.text)
+        return
+
+    if output == 'svg':
+        if filename is None:
+            filename = 'iPath.svg'
+        with open(filename, 'wb') as f:
+            f.write(r.content)
+
+    elif output == 'png':
+        from cairosvg import svg2png
+        if filename is None:
+            filename = 'iPath.png'
+        svg2png(bytestring=r.content, write_to=filename)
+
+    elif output == 'notebook':
+        from IPython.display import SVG, display_svg
+        display_svg(SVG(data=r.content))
+
+    else:
+        print('Output format not supported.')
+
+
+def edge_mapper(data, min_width=0, max_width=25, min_alpha=0.25, max_alpha=1.0):
+
+    if len(data) == 0:
+        return 'empty'
+
+    ub = max(data.values())
+    lb = min(data.values())
+
+    if ub == lb:
+        width_factor = (max_width - min_width) / 2
+        alpha_factor = (max_alpha - min_alpha) / 2
+        lb = 0
+    else:
+        width_factor = (max_width - min_width) / (ub - lb)
+        alpha_factor = (max_alpha - min_alpha) / (ub - lb)
+
+    edges = []
+
+    for id, value in data.items():
+        width = int(ceil((value - lb) * width_factor + min_width))
+        alpha = (value - lb) * alpha_factor + min_alpha
+        edge = '{:s} W{:d} {:.3f}'.format(id, width, alpha)
+        edges.append(edge)
+
+    return edges
+
+
+def display_fluxes(fluxes, id_convert=None, mnx_path=None, log=False, abstol=1e-6, **kwargs):
+
+    if id_convert is not None:
+
+        if mnx_path is None:
+            print('Please specify local path of MetaNetX database.')
+            return
+
+        mnx = MetaNetX(mnx_path, version=3)
+
+        # TODO: still need to find how to deal with ambiguous conversions
+
+        fluxes = {
+            kegg_id: abs(value)
+            for r_id, value in fluxes.items()
+            for kegg_id in mnx.translate_reaction_id(r_id[2:], id_convert, 'kegg')
+        }
+    else:
+        fluxes = {key: abs(val) for key, val in fluxes.items()}
+
+    if log:
+        lb = np.log10(min([x for x in fluxes.values() if x > abstol]))
+        fluxes = {key: np.log10(val) - lb for key, val in fluxes.items() if val > abstol}
+
+    return iPATH_display(fluxes, **kwargs)