Skip to content

Commit

Permalink
fix(rplibs): fix MIRIAM cross-ref reading
Browse files Browse the repository at this point in the history
  • Loading branch information
@breakthewall
breakthewall committed Oct 11, 2023
1 parent 67eed8f commit 070d05b
Show file tree
Hide file tree
Showing 2 changed files with 49 additions and 90 deletions.
91 changes: 34 additions & 57 deletions rptools/rplibs/rpSBML.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2151,7 +2151,8 @@ def speciesMatchWith(
:type compartment_id: str
:type logger: logging
:return: A tuple corresponding to the dictionnary correspondance between species provided and specie in the model and a list of species not find in the model
:return: A tuple corresponding to the dictionnary correspondance between species provided
and specie in the model and a list of species not find in the model
:rtype: Tuple[Dict[str,str], List[str]]
"""
self.logger.debug(f'species_ids: {species_ids}')
Expand Down Expand Up @@ -2246,41 +2247,42 @@ class MatchSpecies(Exception):
try:
# For all (lists of) cross refs in MIRIAM annot
for cross_refs in miriam_annot:
# For all single cross ref in a DB list
for cross_ref in cross_refs:
logger.debug(f'Comparing {species_id_to_match} with {cross_ref} (MIRIAM)')
# Found a match between species ID and the current cross ref
if species_id_to_match == cross_ref:
curr_match = model_species.getId()
curr_score = scores[curr_match]
logger.debug(f'Found match: {species_id_to_match} -> {curr_match} (involved in {curr_score} reactions)')
# Species ID already matched with another species
if species_id_to_match in prev_matches:
prev_match = prev_matches[species_id_to_match]
prev_score = scores[prev_match]
logger.warning(
f'*** A match already exists for {species_id_to_match}'
)
cross_ref = cross_refs.split('/')[-1]
# # For all single cross ref in a DB list
# for cross_ref in cross_refs:
logger.debug(f'Comparing {species_id_to_match} with {cross_ref} (MIRIAM)')
# Found a match between species ID and the current cross ref
if species_id_to_match == cross_ref:
curr_match = model_species.getId()
curr_score = scores[curr_match]
logger.debug(f'Found match: {species_id_to_match} -> {curr_match} (involved in {curr_score} reactions)')
# Species ID already matched with another species
if species_id_to_match in prev_matches:
prev_match = prev_matches[species_id_to_match]
prev_score = scores[prev_match]
logger.warning(
f'*** A match already exists for {species_id_to_match}'
)
logger.warning(
f' \__ previous match: {species_id_to_match} -> {prev_match} (involved in {prev_score} reactions)'
)
logger.warning(
f' \__ current match: {species_id_to_match} -> {curr_match} (involved in {curr_score} reactions)'
)
# Compare the scores of the previous and the current match
if curr_score > prev_score:
logger.warning(
f' \__ previous match: {species_id_to_match} -> {prev_match} (involved in {prev_score} reactions)'
f'--> Keep {species_id_to_match} -> {curr_match}'
)
return curr_match
else:
logger.warning(
f' \__ current match: {species_id_to_match} -> {curr_match} (involved in {curr_score} reactions)'
f'--> Keep {species_id_to_match} -> {prev_match}'
)
# Compare the scores of the previous and the current match
if curr_score > prev_score:
logger.warning(
f'--> Keep {species_id_to_match} -> {curr_match}'
)
return curr_match
else:
logger.warning(
f'--> Keep {species_id_to_match} -> {prev_match}'
)
else:
return curr_match
# Stop the search in the current MIRIAM annotations
raise MatchSpecies
else:
return curr_match
# Stop the search in the current MIRIAM annotations
raise MatchSpecies
except MatchSpecies:
pass

Expand Down Expand Up @@ -2563,32 +2565,7 @@ def _search_key(keys: List[str], dict: Dict) -> str:
return d_reactions


# def __str__(self):
# content = ''
# for attr in inspect_getmembers(self):
# if not attr[0].startswith('_'):
# if not inspect_ismethod(attr[1]):
# # self.logger.info(attr)
# content += f'{attr}\n'
# return content


def __eq__(self, other):
# self_species = self.getModel().getListOfSpecies()
# other_species = other.getModel().getListOfSpecies()
# for rxn in self.getModel().getListOfReactions():
# for elt in rxn.getListOfReactants():
# print(elt.getSpecies(), elt.getStoichiometry(), self_species.get(elt.getSpecies()))
# for elt in rxn.getListOfProducts():
# print(elt.getSpecies(), elt.getStoichiometry(), other_species.get(elt.getSpecies()))
# print(list(rxn.getListOfReactants()))
# print(list(rxn.getListOfProducts()))
# print(list(self.getModel().getListOfReactions()))
# print(list(other.getModel().getListOfReactions()))
# len(self.getModel().getListOfReactions())==len(other.getModel().getListOfReactions()) \
# return \
# sorted(self.readGroupMembers('rp_pathway')) == sorted(other.readGroupMembers('rp_pathway')) \
# and self._get_reactions_with_species_keys(self.logger) == other._get_reactions_with_species_keys(other.logger)
self.logger.debug(f'Comparing {self} and {other}')
return \
self._get_reactions_with_species_keys() \
Expand Down
48 changes: 15 additions & 33 deletions tests/rplibs/test_rpSBML.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -285,13 +285,12 @@ def test_has_reaction(self):

def test_speciesMatchWith(self):
# Return type.
species_match_with = self.rpsbml_none.speciesMatchWith(
corr_spe, miss_spe = self.rpsbml_none.speciesMatchWith(
[],
''
)
self.assertIsInstance(species_match_with, Tuple)
self.assertIsInstance(species_match_with[0], Dict)
self.assertIsInstance(species_match_with[1], List)
self.assertIsInstance(corr_spe, Dict)
self.assertIsInstance(miss_spe, List)
# Basic
self.assertEqual(
sorted(self.rpsbml_lycopene_specie_id),
Expand All @@ -301,46 +300,29 @@ def test_speciesMatchWith(self):
)[0].values())
)
# Challenge - 1
species_match_with = self.rpsbml_lycopene.speciesMatchWith(
['HMDB00250', '13420'],
corr_spe, miss_spe = self.rpsbml_lycopene.speciesMatchWith(
['HMDB00250', 'CHEBI:13420'],
'c'
)
self.assertEqual(
len(species_match_with[0]),
2
self.assertDictEqual(
corr_spe,
{'HMDB00250': 'M_ppi_c', 'CHEBI:13420': 'M_ppi_c'}
)
self.assertEqual(
len(set(species_match_with[0].values())),
1
)
self.assertIn(
'M_ppi_c',
species_match_with[0].values()
)
self.assertFalse(species_match_with[1])
self.assertListEqual(miss_spe, [])
# Challenge - 2
species_match_with = self.rpsbml_lycopene.speciesMatchWith(
corr_spe, miss_spe = self.rpsbml_lycopene.speciesMatchWith(
['HMDB00250', '13420'],
''
)
self.assertFalse(species_match_with[0])
self.assertEqual(
len(species_match_with[1]),
2
)
self.assertDictEqual(corr_spe, {})
self.assertListEqual(sorted(miss_spe), sorted(['HMDB00250', '13420']))
# Challenge - 3
species_match_with = self.rpsbml_lycopene.speciesMatchWith(
corr_spe, miss_spe = self.rpsbml_lycopene.speciesMatchWith(
['MNXM24', '13421'],
'c'
)
self.assertEqual(
['MNXM24'],
list(species_match_with[0].values())
)
self.assertEqual(
['13421'],
species_match_with[1]
)
self.assertDictEqual(corr_spe, {'MNXM24': 'MNXM24'})
self.assertListEqual(miss_spe, ['13421'])

def test_is_boundary_type(self):
# TODO: implement test which doesn't account abount SBO terms, to see how compartment_id ... are managed
Expand Down

0 comments on commit 070d05b

Please sign in to comment.