Skip to content

Commit

Permalink
[pre-commit.ci] auto fixes from pre-commit.com hooks
Browse files Browse the repository at this point in the history
for more information, see https://pre-commit.ci
  • Loading branch information
pre-commit-ci[bot] committed Jun 23, 2022
1 parent 8425db7 commit 92640c6
Show file tree
Hide file tree
Showing 3 changed files with 80 additions and 43 deletions.
4 changes: 3 additions & 1 deletion biomass/exec_model.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,11 @@
import re
from types import ModuleType
from typing import List, NamedTuple
from .graph import NetworkGraph

import numpy as np

from .graph import NetworkGraph


class OptimizedValues(NamedTuple):
params: list
Expand Down
77 changes: 49 additions & 28 deletions biomass/graph.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,10 @@
import os
import re
import warnings
from collections import defaultdict
from types import ModuleType

import pygraphviz as pgv
import re, os, warnings
from collections import defaultdict


class NetworkGraph(object):
Expand All @@ -14,7 +17,7 @@ class NetworkGraph(object):
Path to a biomass model.
biomass_model : Any
A package containing biomass model properties.
Attributes
----------
path : str
Expand Down Expand Up @@ -78,39 +81,53 @@ def _extract_equation(dir: str, filepath: str, left_match: str, right_match: str
with open(os.path.join(dir, filepath)) as f:
was_warned = False
lefthand = None
equation_cache = ''
equation_cache = ""
for line in f.readlines():
match = re.search(left_match, line)
cache_full = bool(len(equation_cache))
if 'return' in line:
if "return" in line:
if cache_full:
equation_cache = ''.join(equation_cache).split('=')[1]
data[lefthand] = [found for found in re.findall(right_match, equation_cache) if found != lefthand]
equation_cache = ''
equation_cache = "".join(equation_cache).split("=")[1]
data[lefthand] = [
found
for found in re.findall(right_match, equation_cache)
if found != lefthand
]
equation_cache = ""
cache_full = False
if match is not None:
if cache_full:
equation_cache = ''.join(equation_cache).split('=')[1]
data[lefthand] = [found for found in re.findall(right_match, equation_cache) if found != lefthand]
equation_cache = ''
equation_cache = "".join(equation_cache).split("=")[1]
data[lefthand] = [
found
for found in re.findall(right_match, equation_cache)
if found != lefthand
]
equation_cache = ""
cache_full = False
lefthand = match.group(0)
equation_cache += line
elif cache_full:
equation_cache += line
if len(equation_cache) == 0:
try:
warning_match = re.findall('\[V\.', line.split('=')[1])
warning_match = re.findall("\[V\.", line.split("=")[1])
except:
warning_match = None
if warning_match and not was_warned:
warnings.warn('Usage of species concentrations outside of rate equations detected.'+
' Species concentrations outside of rate equations will not be considered in Graph generation.'+
' Might lead to faulty graph.')
was_warned=True
warnings.warn(
"Usage of species concentrations outside of rate equations detected."
+ " Species concentrations outside of rate equations will not be considered in Graph generation."
+ " Might lead to faulty graph."
)
was_warned = True
return data

def to_graph(self, save_path: str, gviz_prog: str ='dot', ):
def to_graph(
self,
save_path: str,
gviz_prog: str = "dot",
):
"""
Constructs and draws a directed graph of the model using ode.py/reaction_network.py as text files.
"""
Expand All @@ -123,16 +140,18 @@ def to_graph(self, save_path: str, gviz_prog: str ='dot', ):
use_flux = False

if use_flux == False:
left_re = '(?<=dydt\[V.)(.+?)(?=\])'
right_re = '(?<=y\[V.)(.+?)\]'
edges = self._extract_equation(self.path, 'ode.py', left_re, right_re)
left_re = "(?<=dydt\[V.)(.+?)(?=\])"
right_re = "(?<=y\[V.)(.+?)\]"
edges = self._extract_equation(self.path, "ode.py", left_re, right_re)
elif use_flux == True:
left_re_flux = '(?<=v\[)(.+?)(?=\])'
right_re_flux = '(?<=y\[V.)(.+?)\]'
left_re_ode = '(?<=dydt\[V.)(.+?)(?=\])'
right_re_ode = '(?<=v\[)(.+?)\]'
fluxes = self._extract_equation(self.path, 'reaction_network.py', left_re_flux, right_re_flux)
odes = self._extract_equation(self.path, 'ode.py', left_re_ode, right_re_ode)
left_re_flux = "(?<=v\[)(.+?)(?=\])"
right_re_flux = "(?<=y\[V.)(.+?)\]"
left_re_ode = "(?<=dydt\[V.)(.+?)(?=\])"
right_re_ode = "(?<=v\[)(.+?)\]"
fluxes = self._extract_equation(
self.path, "reaction_network.py", left_re_flux, right_re_flux
)
odes = self._extract_equation(self.path, "ode.py", left_re_ode, right_re_ode)
edges = defaultdict(lambda: [])
for species, velocities in odes.items():
for velocity in velocities:
Expand All @@ -143,7 +162,9 @@ def to_graph(self, save_path: str, gviz_prog: str ='dot', ):
for participants in edges.values():
num_participants += participants
num_participants += edges.keys()
assert set(self.species) == set(num_participants), f"Not all species are extracted. Expected {len(self.species)}, got {len(set(num_participants))}"
assert set(self.species) == set(
num_participants
), f"Not all species are extracted. Expected {len(self.species)}, got {len(set(num_participants))}"
graph = pgv.AGraph(directed=True)
for species, partners in edges.items():
graph.add_node(species)
Expand All @@ -152,4 +173,4 @@ def to_graph(self, save_path: str, gviz_prog: str ='dot', ):
self.graph = graph
graph.layout(prog=gviz_prog)
graph.draw(save_path)
return
return
42 changes: 28 additions & 14 deletions tests/test_graph.py
Original file line number Diff line number Diff line change
@@ -1,21 +1,35 @@
import os, shutil
import os
import shutil

import pytest

from biomass import create_model


def test_graph():
model_lib_list = [name for name in os.listdir('biomass/models') if '__' not in name]
if not os.path.isdir(os.path.join('biomass/models', '__graphs')):
os.mkdir(os.path.join('biomass/models', '__graphs'))
model_lib_list = [name for name in os.listdir("biomass/models") if "__" not in name]
if not os.path.isdir(os.path.join("biomass/models", "__graphs")):
os.mkdir(os.path.join("biomass/models", "__graphs"))
for model_name in model_lib_list:
if model_name in ['prolif_quies', 'g1s_transition']:
model = create_model('.'.join(('biomass', 'models', model_name)))
if model_name in ["prolif_quies", "g1s_transition"]:
model = create_model(".".join(("biomass", "models", model_name)))
with pytest.warns(UserWarning):
model.to_graph(os.path.join('biomass/models', '__graphs', model_name + 'test.png'))
assert os.stat(os.path.join('biomass/models', '__graphs', model_name + 'test.png')).st_size > 1024*10
model.to_graph(os.path.join("biomass/models", "__graphs", model_name + "test.png"))
assert (
os.stat(
os.path.join("biomass/models", "__graphs", model_name + "test.png")
).st_size
> 1024 * 10
)
continue
model = create_model('.'.join(('biomass', 'models', model_name)))
model.to_graph(os.path.join('biomass/models', '__graphs', model_name + 'test.png'))
assert os.stat(os.path.join('biomass/models', '__graphs', model_name + 'test.png')).st_size > 1024*10

model = create_model(".".join(("biomass", "models", model_name)))
model.to_graph(os.path.join("biomass/models", "__graphs", model_name + "test.png"))
assert (
os.stat(os.path.join("biomass/models", "__graphs", model_name + "test.png")).st_size
> 1024 * 10
)


def test_cleanup():
if os.path.isdir(os.path.join('biomass/models', '__graphs')):
shutil.rmtree(os.path.join('biomass/models', '__graphs'))
if os.path.isdir(os.path.join("biomass/models", "__graphs")):
shutil.rmtree(os.path.join("biomass/models", "__graphs"))

0 comments on commit 92640c6

Please sign in to comment.