Visualizes information flow connections in humans
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 

331 lines
11 KiB

#!/bin/python
# This script will generate a graphviz dotfile from the specified dataset.
# Generally it should be called from build.sh only.
#
# The behavior can be influenced by setting certain environment variables:
# - COLLAPSE={space-separated list of nodes}
# - NO_COLLAPSE={space-separated list of nodes}
# - DELETE_NODES={space-separated list of nodes}
# - GREEDY_COLLAPSE={1,0}
# - DEDUPLICATE_EDGES={1,0}
# - USE_CLUSTERS={1,0}
# - GENERATE_GROUP_CONNECTIONS={1,0}
# - DELETE_LONE_NODES={1,0}
# - FULL_NAMES={1,0}
import csv, sys, os
if len(sys.argv) != 4:
print("usage: convert.sh <nodes.csv> <edges.csv> <functions.csv>")
sys.exit()
nodereader = csv.reader(open(sys.argv[1], 'r'), skipinitialspace=True)
edgereader = csv.reader(open(sys.argv[2], 'r'), skipinitialspace=True)
functionreader = csv.reader(open(sys.argv[3], 'r'), skipinitialspace=True)
nodetypes = {
'source': 'rank=min shape=rarrow',
'sink': 'rank=max',
'group': 'shape=box penwidth=2',
'complex': 'penwidth=2',
'relay': 'fillcolor="#ffff55"',
'null': 'style=invis',
'muscle': 'color=red',
}
edgetypes = {
'GABA': 'color=red tooltip=GABA arrowhead=tee',
'Glutamate': 'color=purple tooltip=Glutamate',
'Serotonin': 'color=blue tooltip=Serotonin',
'Dopamine': 'color=green tooltip=Dopamine',
'ACh': 'color=orange tooltip=Acetylcholine',
'Adrenergic': 'color=red tooltip="(Nor)Adrenaline"',
'Melatonin': 'color="#cc0033" tooltip=Melatonin',
'none': 'style=invis',
'weak': 'style=dashed tooltip=weak weight=0',
'strong': 'penwidth=2 tooltip=strong weight=10',
'vstrong': 'penwidth=3 tooltip="very strong" weight=20',
'vvstrong': 'penwidth=4 tooltip="super strong" weight=30',
'via': 'color="#999999" tooltip="transitively connected"',
'constitutes': 'arrowhead=none arrowtail=none style="dotted" arrowsize=1.5 color="#000000" dir=both penwidth=1 tooltip=constitutes',
'nodir': 'arrowhead=none',
'related': 'style=invis',
'dubious': '',
'bi': 'arrowtail=normal',
'anion': '',
'cation': 'color="#009944" tooltip="Depolarization by influx of cations"',
'vibration': 'arrowhead=normal style="dotted,setlinewidth(7)" arrowsize=1 tooltip="Vibrations"',
'G-protein': 'color="#228844" tooltip="Intracellular"',
}
groupcolors = {
'cns.spinal': '#ffff55',
'cns.nerve': '#ffff55',
'cns.retina': '#99ff99',
'body.face.eye': '#99ff99',
'cns': '#9999ff',
'cns.vis': '#00ff00',
'cns.basal.thal': '#ffbbff',
'cns.basal': '#ffcc99',
'cns.brainstem': '#ff9999',
'cns.frontal': '#99ccff',
'cns.pari': '#ff55ff',
'cns.temp': '#00ffcc',
'cns.epithal': '#009090',
'cns.olfactory': '#6666ff',
'body.hand': '#ffeecc',
'cns.cer': '#cc0000',
'body.face.ear': '#bbaa77',
'body.face.tongue': '#ff0077',
'lh': '#ff9999',
'rh': '#9999ff',
}
DOT_HEADER = """
digraph brainvis {
overlap=false;
graph [dpi=72 rankdir="LR" fontsize=14 fontname="mono bold" style="rounded,filled,setlinewidth(0)" fillcolor="#aaaaaa" compound=True];
node [shape=egg fontsize=10, fontname=mono style=filled fillcolor="#dddddd"];
edge [fontsize=10, fontname=mono arrowhead=vee];
"""
collapse = os.environ.get('COLLAPSE', '').split()
no_collapse = os.environ.get('NO_COLLAPSE', '').split()
delete_nodes = os.environ.get('DELETE_NODES', '').split()
greedy_collapse = bool(int(os.environ.get('GREEDY_COLLAPSE', '0')))
deduplicate_edges = bool(int(os.environ.get('DEDUPLICATE_EDGES', '1')))
use_clusters = bool(int(os.environ.get('USE_CLUSTERS', '0')))
delete_nodes_without_connections = bool(int(os.environ.get('DELETE_LONE_NODES', '0')))
generate_group_connections = bool(int(os.environ.get('GENERATE_GROUP_CONNECTIONS', '0')))
invisible_group_connections = bool(int(os.environ.get('INVISIBLE_GROUP_CONNECTIONS', '1')))
full_names = bool(int(os.environ.get('FULL_NAMES', '0')))
if not invisible_group_connections:
edgetypes['related'] = 'arrowhead=none color="#999999"'
class Node(object):
def __init__(self, row):
self.token = row[0]
self.name = row[1]
self.fullname = row[2] or self.name
self.attr = row[3]
self.uri = row[4]
self._functiontext = []
self._functionuris = []
def addfunction(self, function, uri):
self._functiontext.append(function)
self._functionuris.append(uri)
@property
def functions(self):
return "".join("\n- %s" % fnc for fnc in self._functiontext)
@property
def tooltip(self):
return "%s (%s)%s" % (self.fullname, self.token, self.functions)
def __map__(self):
return {'token': self.token,
'name': self.name,
'fullname': self.fullname,
'uri': self.uri,
'tooltip': self.tooltip}
class Edge(object):
def __init__(self, row):
self.source = row[0]
self.target = row[1]
self.attr= row[2]
self.uri = row[3]
if ' ' in self.uri:
self.uri = self.uri.split()[0]
def __str__(self):
return "%s, %s" % (self.source, self.target)
def __map__(self):
return {'source': self.source,
'target': self.target,
'uri': self.uri}
edges = []
nodes = {}
groups = {}
subgraphs = {}
subgraph_groups = {}
functions = {}
# Collect data
for row in nodereader:
if not row:
continue
if len(row) > 5:
sys.stderr.write("Warning: too many values for node %s\n" % str(row))
try:
node = Node(row)
except:
sys.stderr.write("Warning: unable to handle node %s\n" % str(row))
else:
nodes[node.token] = node
if 'group' in node.attr or 'complex' in node.attr:
groups[node.token] = node.token
if use_clusters:
if 'subgraph' in node.attr:
subgraphs[node.token] = node
subgraph_groups[node.token] = [node.token]
for row in edgereader:
if not row:
continue
try:
edge = Edge(row)
except:
sys.stderr.write("Warning: unable to handle edge %s\n" % str(row))
else:
edges.append(edge)
for row in functionreader:
if not row:
continue
try:
token, function, uri = row
except:
sys.stderr.write("Warning: unable to handle function %s\n" % str(row))
else:
try:
node = nodes[token]
except KeyError:
sys.stderr.write("Warning: cannot assign function to %s, no such node. Skipping.\n" % token)
else:
node.addfunction(function, uri)
# Generate group connections
grouptokens = list(reversed(sorted(groups)))
nodetokens = sorted(nodes)
if generate_group_connections:
for node in nodetokens:
for group in grouptokens:
if node.startswith(group) and node != group:
#sys.stderr.write("connecting %s with %s\n" % (group, node))
edges.append(Edge([group, node, 'related', '']))
break
edges.append(Edge(['body', 'cns', 'related', '']))
# Determine subgraph_groups
grouptokens = list(reversed(sorted(subgraphs)))
if use_clusters:
for node in sorted(nodes):
for group in grouptokens:
if node.startswith(group):
subgraph_groups[group].append(node)
# Collapse nodes into a collective group node
#sys.stderr.write(repr(sorted(collapse)))
for group in sorted(collapse) if greedy_collapse else reversed(sorted(collapse)):
if not group:
continue
# sys.stderr.write("group: %s\n" % group)
groupnodes = []
groupnodenames = []
for node in nodes.values():
if node.token.startswith(group) and \
(node.token != group if greedy_collapse else node.token not in collapse) and \
not any(node.token.startswith(exception) for exception in no_collapse):
# sys.stderr.write("node: %s\n" % node.token)
groupnodes.append(node)
groupnodenames.append(node.token)
for node in groupnodes:
del nodes[node.token]
i = len(edges)
while i > 0:
i -= 1
edge = edges[i]
if edge.source in groupnodenames:
edge.source = group
if edge.target in groupnodenames:
edge.target = group
if edge.source == group and edge.target == group:
del edges[i]
# Deduplicate edges
if deduplicate_edges:
unique_edges = set()
i = len(edges)
for edge in reversed(list(edges)):
i -= 1
connection = (edge.source, edge.target)
if connection in unique_edges:
del edges[i]
else:
unique_edges.add(connection)
# Generate groups (for dot only)
#for node in nodes:
# name, fullname, group, type_, uri = node
# if group:
# if group in groups:
# groups[group].append(name)
# else:
# groups[group] = [name]
for token in delete_nodes:
if token in nodes:
del nodes[token]
edges = [edge for edge in edges if edge.target not in delete_nodes and \
edge.source not in delete_nodes]
if delete_nodes_without_connections:
connected_nodes = set([edge.source for edge in edges]) | set([edge.target for edge in edges])
if generate_group_connections:
connected_nodes |= set(groups)
nodes = dict((k, v) for k, v in nodes.items() if k in connected_nodes)
# Format data
print(DOT_HEADER)
if use_clusters:
for name, items in sorted(subgraph_groups.items()):
print("""
subgraph "cluster_{0}" {{
label = "{0}";
margin = 3;
{1}
}}\n""".format(name, ", ".join('"%s"' % item for item in items)))
for token in sorted(nodes):
node = nodes[token]
properties = []
closest_color = ""
for key in groupcolors:
if len(key) > len(closest_color) and node.token.startswith(key):
closest_color = key
if closest_color:
properties.append('fillcolor="%s"' % groupcolors[closest_color])
properties += [nodetypes.get(type_, "") for type_ in node.attr.split('|')]
if full_names:
node.name = node.fullname
print(' "{token}" [{properties} label="{name}" URL="{uri}" tooltip="{tooltip}"];'.format(properties=" ".join(properties), **node.__map__()))
for edge in sorted(edges, key=str):
attrs = []
if use_clusters:
if edge.source in subgraphs:
attrs.append('ltail="cluster_{token}"'.format(**subgraphs[edge.source].__map__()))
if edge.target in subgraphs:
attrs.append('lhead="cluster_{token}"'.format(**subgraphs[edge.target].__map__()))
for type_ in edge.attr.split("|"):
if type_ == 'none':
continue
attrs.append(edgetypes.get(type_, ""))
attrs = " ".join(attrs)
print(' "{source}" -> "{target}" [URL="{uri}" {attrs}];'.format(attrs=attrs, **edge.__map__()))
print("}")