Skip to content
12 changes: 6 additions & 6 deletions src/openrxn/compartments/ID.py
Original file line number Diff line number Diff line change
@@ -1,16 +1,16 @@

def join_tup(tup):
def join_tuple(tup):
return '_'.join([str(a) for a in tup])

def makeID(array_ID,comp_ID):
def make_ID(array_ID,comp_ID):
if array_ID is not None:
tag = array_ID + '-'
else:
tag = ""

if type(comp_ID) is tuple or type(comp_ID) is list:
return tag + join_tup(comp_ID)
elif type(comp_ID) is str:
if isinstance(comp_ID, tuple) or isinstance(comp_ID, list):
return tag + join_tuple(comp_ID)
elif isinstance(comp_ID, str):
return tag + comp_ID
elif type(comp_ID) is int:
elif isinstance(comp_ID, int):
return tag + str(comp_ID)
39 changes: 26 additions & 13 deletions src/openrxn/compartments/compartment.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Compartments hold a set of reactions and govern the transport
of material between compartments through connections."""

from openrxn.compartments.ID import makeID
from openrxn.compartments.ID import make_ID
from openrxn import unit
import logging

Expand All @@ -22,20 +22,22 @@ class Compartment(object):
Upon initialization, both of these lists are empty.
"""

def __init__(self, ID, pos=[], array_ID=None, volume=None):
def __init__(self, ID, pos=None, array_ID=None, volume=None):
self.ID = ID
self._rxn_ids = set()
self.reactions = []
self.connections = {}
self.pos = pos
self.pos = [] if pos is None else list(pos)
self.array_ID = array_ID
self.volume = volume

def add_rxn_to_compartment(self, rxn):
"""Adds a reaction to a compartment."""
if rxn.ID in [r.ID for r in self.reactions]:
logging.warn("Reaction {0} already in compartment {1}".format(rxn.ID,self.ID))
if rxn.ID in self._rxn_ids:
logging.warning("Reaction {0} already in compartment {1}".format(rxn.ID,self.ID))
else:
self.reactions.append(rxn)
self._rxn_ids.add(rxn.ID)

def add_rxns_to_compartment(self, rxns):
"""Adds a list of reactions to a compartment."""
Expand All @@ -54,20 +56,20 @@ def connect(self, other_compartment, conn_type, warn_overwrite=True):
"""Make a connection from this compartment to another one
using the conn_type connection type."""

conn_tag = makeID(other_compartment.array_ID,other_compartment.ID)
conn_tag = make_ID(other_compartment.array_ID,other_compartment.ID)
if conn_tag in self.connections and warn_overwrite:
self_tag = makeID(self.array_ID,self.ID)
logging.warn("Warning: overwriting connection between {0} and {1}".format(self_tag,conn_tag))
self_tag = make_ID(self.array_ID,self.ID)
logging.warning("Overwriting connection between {0} and {1}".format(self_tag,conn_tag))

self.connections[conn_tag] = (other_compartment, conn_type)

def remove_connection(self, other_compartment):
"""Remove the connection with the other_compartment"""

conn_tag = makeID(other_compartment.array_ID,other_compartment.ID)
conn_tag = make_ID(other_compartment.array_ID,other_compartment.ID)
if conn_tag not in self.connections:
self_tag = makeID(self.array_ID,self.ID)
logging.warn("Warning: connection to remove between {0} and {1} does not exist".format(self_tag,conn_tag))
self_tag = make_ID(self.array_ID,self.ID)
logging.warning("Connection to remove between {0} and {1} does not exist".format(self_tag,conn_tag))

val = self.connections.pop(conn_tag)

Expand All @@ -89,11 +91,22 @@ def copy(self,ID=None,delete_array_ID=False):
new_comp = type(self)(newID, pos=self.pos, array_ID=new_aID)

new_comp.volume = self.volume
new_comp.connections = self.connections
new_comp.reactions = self.reactions
new_comp.connections = dict(self.connections)
new_comp.reactions = list(self.reactions)
new_comp._rxn_ids = {r.ID for r in new_comp.reactions}

return new_comp

def __repr__(self):
vol_str = str(self.volume) if self.volume is not None else "None"
pos_str = tuple(self.pos) if self.pos else None

return (
f"Compartment(ID={self.ID!r}, "
f"volume={vol_str}, pos={pos_str}, "
f"n_rxns={len(self.reactions)}, n_connections={len(self.connections)})"
)

class Compartment1D(Compartment):

def __init__(self, *args, **kwargs):
Expand Down
168 changes: 124 additions & 44 deletions src/openrxn/connections.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
"""Connections govern the transport between compartments.
"""
Connections govern the transport between compartments.
In some cases the transport can be described by a first-order
rate equation, e.g.:

Expand All @@ -15,6 +16,7 @@
"""

from . import unit
import logging

class Connection(object):
"""Basis class for connections. The argument species_rates
Expand All @@ -24,72 +26,126 @@ class Connection(object):
class AnisotropicConnection(Connection):

def __init__(self, species_rates,dim=3):
"""AnisotropicConnections are initialized with a dictionary
of species_rates, where the keys are Species IDs and the
values are tuples of transport rates (k_out,k_in).
"""
Directional (anisotropic) transport connection between two compartments.

Parameters
----------
species_rates : dict
Dictionary of Species, where the keys are Species IDs and the
values are tuples of transport rates (k_out,k_in).
k_out corresponds to transport from compartment 1 to 2.
k_in corresponds to transport from compartment 2 to 1.

- A tuple (k_out, k_in) specifies directional transport rates.
- A scalar quantity k is interpreted as symmetric transport
and internally turn into tuples of transport rates (k, k).
- All rates must be specified in units of 1/s.

Care should be taken to make sure these are applied in the
right direction!
dim : int, optional
Spatial dimension.

Rates should be specified in units of 1/s.
"""
self.species_rates = species_rates
self.dim = dim

for s,r in self.species_rates.items():
if not isinstance(r,tuple):
logging.warning(f"Species {s}: one scalar rate provided. Assigning k_out == k_in.")
r = (r,r)
self.species_rates[s] = r

elif len(r) != 2:
raise ValueError(f"Species {s}: rate tuple must be length 2, got length {len(r)}.")

for s in self.species_rates:
if type(self.species_rates[s]) is not tuple or len(self.species_rates[s]) != 2:
raise ValueError("Error! Elements of species_rates dictionary should be tuples of length 2")
self.species_rates[s][0].ito(1/unit.sec)
self.species_rates[s][1].ito(1/unit.sec)

@staticmethod
def _flip_tuple(t):
return (t[1],t[0])

def reverse(self):
rev_species_rates = {}
for s in self.species_rates:
rev_species_rates[s] = self._flip_tuple(self.species_rates[s])
for s,r in self.species_rates.items():
rev_species_rates[s] = self._flip_tuple(r)

return AnisotropicConnection(rev_species_rates)


def __repr__(self):
rates = {s: (k[0].magnitude, k[1].magnitude) for s, k in self.species_rates.items()}
unit_str = str(next(iter(self.species_rates.values()))[0].units)
return (f"AnisotropicConnection(dim={self.dim},"
f"n_species={len(self.species_rates)},"
f"transport_rates={rates}{unit_str})"
)

class IsotropicConnection(Connection):

def __init__(self, species_rates,dim=3):
"""IsotropicConnections are initialized with a dictionary
of species_rates, where the keys are Species IDs and the
values are transport rates.
"""
Symmetric (isotropic) transport connection between two compartments.

Parameters
----------
species_rates : dict
Dictionary of Species, where the keys are Species IDs and the
value is a transport rate constant (k).

- A scalar quantity k is interpreted as symmetric transport
and internally turn into tuples of transport rates (k, k).
- All rates must be convertible to units of 1/s.

dim : int, optional
Spatial dimension.

Rates should be specified in units of 1/s.
"""
self.species_rates = species_rates
self.dim = dim

for s in self.species_rates:
k = self.species_rates[s]
if type(k) is not tuple:
for s,k in self.species_rates.items():
if not isinstance(k,tuple):
self.species_rates[s] = (k,k)

self.species_rates[s][0].ito(1/unit.sec)
self.species_rates[s][1].ito(1/unit.sec)

def __repr__(self):
rates = {s: (k[0].magnitude, k[1].magnitude) for s, k in self.species_rates.items()}
unit_str = str(next(iter(self.species_rates.values()))[0].units)
return (f"IsotropicConnection(dim={self.dim},"
f"n_species={len(self.species_rates)},"
f"transport_rates={rates}{unit_str})"
)

class DivByVConnection(Connection):

def __init__(self, species_rates,dim=3):
"""DivByVConnections are initialized with a dictionary
of species_rates, where the keys are Species IDs and the
values are transport rates.
"""
Volume-based transport connection between two compartments.

This connection stores transport coefficients proportional to compartment volume.
Rates should be specified in units of L^d/s, where L is length and d is the
compartment volume.

Rates should be specified in units of L^d/s, where L is length
and d is the compartment volume.
Parameters
----------
species_rates : dict
Dictionary of Species, where the keys are Species IDs and the
value is a transport rate constant (k).

dim : int, optional
Spatial dimension.

These connections are divided by the compartment volume
when constructing a system.
"""

self.species_rates = species_rates
self.dim = dim

for s in self.species_rates:
k = self.species_rates[s]
if type(k) is not tuple:
self.species_rates[s] = (k,k)
for s,k in self.species_rates.items():
if not isinstance(k, tuple):
k = (k,k)
self.species_rates[s] = k
self.species_rates[s][0].ito(unit.nm**self.dim/unit.sec)
self.species_rates[s][1].ito(unit.nm**self.dim/unit.sec)

Expand Down Expand Up @@ -127,6 +183,19 @@ def __init__(self, species_d_constants, surface_area=None, ic_distance=None, dim

If either surface_area or ic_distance is left undefined, they will be
automatically calculated using compartment positions.

Parameters
----------
species_d_constants : dict
Dictionary of Species, where the keys are Species IDs and the
value is a diffusion constant (D). (units convertible to L^2 / s).
surface_area : Quantity, optional
Interface area A between compartments (units convertible to L^(dim-1)).
ic_distance : Quantity, optional
Center-to-center distance DeltaX (units convertible to L).
dim : int, optional
Spatial dimension (default 3).

"""

self.species_d_constants = species_d_constants
Expand All @@ -139,15 +208,13 @@ def resolve(self):
require any information about the Species, or the arrays"""

if self.surface_area is None or self.ic_distance is None:
raise ValueError("Error! This connection is not ready to be resolved.")
species_list = self.species_d_constants.keys()
raise ValueError("Error! This connection is not ready to be resolved.")

rates = {}
for s,d in self.species_d_constants.items():

# how are you going to get this volume?

rates[s] = d*self.surface_area/self.ic_distance
rates[s].ito(unit.nm**self.dim/unit.sec)
kV = d*self.surface_area/self.ic_distance
kV.ito(unit.nm**self.dim/unit.sec)
rates[s] = kV

return DivByVConnection(rates,self.dim)

Expand All @@ -158,8 +225,20 @@ def __init__(self, species_d_constants, surface_area=None, ic_distance=None, dim
ResConnections are special connections from a compartment to a
reservoir. They are resolved similarly to a FicksConnection.

face is a string input, equal to either 'x', 'y' or 'z',
denoting along which axis the ResConnection takes place.
Parameters
----------
species_d_constants : dict
Dictionary of Species, where the keys are Species IDs and the
value is a diffusion constant (D). (units convertible to L^2/s).
surface_area : Quantity, optional
Reservoir interface area A (units convertible to L^(dim-1)).
ic_distance : Quantity, optional
Distance DeltaX from compartment center to reservoir boundary (units convertible to L).
dim : int, optional
Spatial dimension.
face : str({'x','y','z'}) or None, optional
Axis normal to the reservoir interface.

"""

self.species_d_constants = species_d_constants
Expand All @@ -173,11 +252,12 @@ def resolve(self):
require any information about the Species, or the arrays"""

if self.surface_area is None or self.ic_distance is None:
raise ValueError("Error! This connection is not ready to be resolved.")
species_list = self.species_d_constants.keys()
raise ValueError("Error! This connection is not ready to be resolved.")

rates = {}
for s,d in self.species_d_constants.items():
rates[s] = d*self.surface_area/self.ic_distance
rates[s].ito(unit.nm**self.dim/unit.sec)
kV = d*self.surface_area/self.ic_distance
kV.ito(unit.nm**self.dim/unit.sec)
rates[s] = kV

return DivByVConnection(rates,self.dim)
8 changes: 4 additions & 4 deletions src/openrxn/model.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@

from openrxn.compartments.compartment import Compartment
from openrxn.reactions import Reaction
from openrxn.compartments.ID import makeID
from openrxn.compartments.ID import make_ID
from openrxn.connections import FicksConnection, ResConnection

import numpy as np
Expand Down Expand Up @@ -205,7 +205,7 @@ def n_compartments(self):
return len(self.compartments)

def add_compartment(self,compartment):
newID = makeID(compartment.array_ID,compartment.ID)
newID = make_ID(compartment.array_ID,compartment.ID)
if newID in self.compartments.keys():
raise ValueError("Error! Duplicate compartment ID in model ({0})".format(newID))
self.compartments[newID] = compartment.copy(ID=newID,delete_array_ID=True)
Expand Down Expand Up @@ -251,14 +251,14 @@ def to_graph(self,scale=10):
# add all the nodes
for c_name, c in self.compartments.items():
graph.add_node(c_name)
graph.node[c_name]['viz'] = {}
graph.nodes[c_name]['viz'] = {}

x = scale*0.5*(c.pos[0][0] + c.pos[0][1]).magnitude
y = scale*0.5*(c.pos[1][0] + c.pos[1][1]).magnitude
z = scale*0.5*(c.pos[2][0] + c.pos[2][1]).magnitude
vis_x,vis_y = self._project_xy((x,y,z))

graph.node[c_name]['viz']['position'] = {'x': float(vis_x), 'y': float(vis_y)}
graph.nodes[c_name]['viz']['position'] = {'x': float(vis_x), 'y': float(vis_y)}

# build an edges list
edges = []
Expand Down
Loading