Adds the pathway organizer back to the project

Author: erikhuck

.gitignore

@@ -1,5 +1,6 @@
 .coverage
 .env/
+.pypi-env/
 .idea/
 __pycache__/
 src/kegg_pull.egg-info/

dev/test_main.py

@@ -9,6 +9,7 @@
 import kegg_pull.rest_cli as r_cli
 import kegg_pull.pull_cli as p_cli
 import kegg_pull.map_cli as map_cli
+import kegg_pull.pathway_organizer_cli as po_cli
 import dev.utils as u
 
 
@@ -19,7 +20,7 @@ def test_help(mocker):
     delimiter: str = '-'*80
     expected_print_call_args = [
         (m.__doc__,), (delimiter,), (p_cli.__doc__,), (delimiter,), (ei_cli.__doc__,), (delimiter,), (map_cli.__doc__,),
-        (delimiter,), (r_cli.__doc__,)]
+        (delimiter,), (po_cli.__doc__,), (delimiter,), (r_cli.__doc__,)]
     u.assert_call_args(function_mock=print_mock, expected_call_args_list=expected_print_call_args, do_kwargs=False)
     for help_arg in (['--help'], ['-h'], []):
         help_args = ['kegg_pull']
@@ -198,3 +199,10 @@ def test_map(mocker, print_output: bool, args: list, stdin_mock_str: str, expect
         json_output=True)
     if stdin_mock:
         stdin_mock.assert_called_once_with()
+
+
+def test_pathway_organizer(mocker, print_output: bool):
+    args = ['kegg_pull', 'pathway-organizer', '--tln=Metabolism', '--fn=Global and overview maps']
+    _test_output(
+        mocker=mocker, args=args, expected_output='dev/test_data/pathway-organizer/metabolic-pathways.json',
+        print_output=print_output, json_output=True)

dev/test_pathway_organizer.py

@@ -0,0 +1,100 @@
+# noinspection PyPackageRequirements
+import pytest as pt
+import json
+import typing as t
+import kegg_pull.pathway_organizer as po
+import dev.utils as u
+
+
+def test_load_from_kegg_warning(mocker, caplog):
+    get_mock: mocker.MagicMock = _get_get_mock(mocker=mocker)
+    parse_hierarchy_spy: mocker.MagicMock = mocker.spy(po.PathwayOrganizer, '_parse_hierarchy')
+    pathway_org: po.PathwayOrganizer = po.PathwayOrganizer.load_from_kegg(top_level_nodes={'invalid-top-level-node'})
+    get_mock.assert_called_once_with(entry_ids=['br:br08901'], entry_field='json')
+    u.assert_warning(
+        message='Top level node name "invalid-top-level-node" is not recognized and will be ignored. Valid values are: "Cellular '
+                'Processes, Drug Development, Environmental Information Processing, Genetic Information Processing, '
+                'Human Diseases, Metabolism, Organismal Systems"', caplog=caplog)
+    parse_hierarchy_spy.assert_called_once_with(pathway_org, level=1, raw_hierarchy_nodes=[], parent_name=None)
+    assert pathway_org.hierarchy_nodes == dict()
+
+
+def _get_get_mock(mocker):
+    def get_mock(**_) -> mocker.MagicMock:
+        with open('dev/test_data/pathway-organizer/pathway-hierarchy.json', 'r') as file_:
+            text_body_mock: str = file_.read()
+        kegg_response_mock = mocker.MagicMock(text_body=text_body_mock)
+        return kegg_response_mock
+    return mocker.patch('kegg_pull.pathway_organizer.r.KEGGrest.get', wraps=get_mock)
+
+
+test_load_from_kegg_data = [
+    (None, None, 'all-nodes.json'),
+    ({'Metabolism', 'Genetic Information Processing'}, None, 'top-level-nodes.json'),
+    (None, {'Genetic Information Processing', 'Global and overview maps', '00010  Glycolysis / Gluconeogenesis'}, 'filter-nodes.json')]
+
+
+@pt.mark.parametrize('top_level_nodes,filter_nodes,hierarchy_nodes_file', test_load_from_kegg_data)
+def test_load_from_kegg(mocker, top_level_nodes: set, filter_nodes: set, hierarchy_nodes_file: str):
+    get_mock: mocker.MagicMock = _get_get_mock(mocker=mocker)
+    pathway_organizer = po.PathwayOrganizer.load_from_kegg(top_level_nodes=top_level_nodes, filter_nodes=filter_nodes)
+    get_mock.assert_called_once_with(entry_ids=['br:br08901'], entry_field='json')
+    if top_level_nodes is not None:
+        actual_top_level_nodes = {node_key for node_key, node_val in pathway_organizer.hierarchy_nodes.items() if node_val['level'] == 1}
+        assert actual_top_level_nodes == top_level_nodes
+    if filter_nodes is not None:
+        for filter_node in filter_nodes:
+            assert filter_node not in pathway_organizer.hierarchy_nodes.keys()
+    expected_hierarchy_nodes: dict = _get_expected_hierarchy_nodes(hierarchy_nodes_file=hierarchy_nodes_file)
+    assert pathway_organizer.hierarchy_nodes == expected_hierarchy_nodes
+
+
+def _get_expected_hierarchy_nodes(hierarchy_nodes_file: str) -> dict:
+    with open(f'dev/test_data/pathway-organizer/{hierarchy_nodes_file}') as file:
+        expected_hierarchy_nodes: dict = json.load(file)
+    return expected_hierarchy_nodes
+
+
+def test_save_to_json(mocker, json_file_path: str):
+    u.mock_non_instantiable(mocker=mocker)
+    pathway_organizer = po.PathwayOrganizer()
+    pathway_organizer.hierarchy_nodes = _get_expected_hierarchy_nodes(hierarchy_nodes_file='top-level-nodes.json')
+    pathway_organizer.save_to_json(file_path=json_file_path)
+    u.test_save_to_json(json_file_path=json_file_path, expected_saved_json_object=pathway_organizer.hierarchy_nodes)
+
+
+def test_load_from_json(json_file_path: str):
+    expected_hierarchy_nodes: dict = _get_expected_hierarchy_nodes(hierarchy_nodes_file='top-level-nodes.json')
+    u.test_load_from_json(
+        json_file_path=json_file_path, saved_object=expected_hierarchy_nodes, method=po.PathwayOrganizer.load_from_json,
+        expected_loaded_object=expected_hierarchy_nodes, loaded_object_attribute='hierarchy_nodes')
+
+
+test_invalid_load_from_json_data = [
+    1, 'a', [], [1, 2], ['a', 'b'], [[], []], [[1], [2]], [['a'], ['b']], [{}, {}], [{'a': {}, 'b': []}], {}, {'a': []}, {'a': {}},
+    {'a': {'b': 1}}, {'a': {'name': 'b'}}, {'a': {'level': 1, 'b': 'c'}},
+    {'a': {'name': 'b', 'level': 1, 'parent': 'c', 'children': None, 'entry_id': 'x'},
+     '': {'name': 'b', 'level': 1, 'parent': 'c', 'children': ['d'], 'entry_id': None}},
+    {'a': {'name': 'b', 'level': 1, 'parent': 'c', 'children': None, 'entry_id': None, 'x': 'y'}},
+    {'a': {'name': 2, 'level': 1, 'parent': 'c', 'children': None, 'entry_id': None}},
+    {'a': {'name': '', 'level': 1, 'parent': 'c', 'children': None, 'entry_id': None}},
+    {'a': {'name': None, 'level': 1, 'parent': 'c', 'children': None, 'entry_id': None}},
+    {'a': {'name': 'b', 'level': '1', 'parent': 'c', 'children': None, 'entry_id': None}},
+    {'a': {'name': 'b', 'level': None, 'parent': 'c', 'children': None, 'entry_id': None}},
+    {'a': {'name': 'b', 'level': 0, 'parent': 'c', 'children': None, 'entry_id': None}},
+    {'a': {'name': 'b', 'level': 1, 'parent': '', 'children': None, 'entry_id': None}},
+    {'a': {'name': 'b', 'level': 1, 'parent': 2, 'children': None, 'entry_id': None}},
+    {'a': {'name': 'b', 'level': 1, 'parent': 'c', 'children': [], 'entry_id': None}},
+    {'a': {'name': 'b', 'level': 1, 'parent': 'c', 'children': [1], 'entry_id': None}},
+    {'a': {'name': 'b', 'level': 1, 'parent': 'c', 'children': [''], 'entry_id': None}},
+    {'a': {'name': 'b', 'level': 1, 'parent': 'c', 'children': ['a'], 'entry_id': 1}},
+    {'a': {'name': 'b', 'level': 1, 'parent': 'c', 'children': ['a'], 'entry_id': ''}}]
+
+
+@pt.mark.parametrize('invalid_json_object', test_invalid_load_from_json_data)
+def test_invalid_load_from_json(caplog, json_file_path: str, invalid_json_object: list | dict | int | float | str):
+    expected_error_message = f'Failed to load the hierarchy nodes. The pathway organizer JSON file at {json_file_path} is ' \
+                             f'corrupted and will need to be re-created.'
+    u.test_invalid_load_from_json(
+        json_file_path=json_file_path, invalid_json_object=invalid_json_object, method=po.PathwayOrganizer.load_from_json,
+        expected_error_message=expected_error_message, caplog=caplog)

dev/test_pathway_organizer_cli.py

@@ -0,0 +1,58 @@
+# noinspection PyPackageRequirements
+import pytest as pt
+import json
+import kegg_pull.pathway_organizer as po
+import kegg_pull.pathway_organizer_cli as po_cli
+import dev.utils as u
+
+
+def test_help(mocker):
+    u.assert_help(mocker=mocker, module=po_cli, subcommand='pathway-organizer')
+
+
+method = 'pathway_organizer_cli.po.PathwayOrganizer.load_from_kegg'
+test_data = [
+    (['pathway-organizer', '--tln=-', '--fn=-'], {'top_level_nodes': {'node1'}, 'filter_nodes': {'node2', 'node3'}},
+     ' node1\n---\nnode2\t\nnode3 '),
+    (['pathway-organizer', '--tln=-', '--fn=node2,node3,node4'],
+     {'top_level_nodes': {'node1', 'node5'}, 'filter_nodes': {'node2', 'node3', 'node4'}}, '\nnode1\n node5\n'),
+    (['pathway-organizer', '--tln=node1', '--fn=-'], {'top_level_nodes': {'node1'}, 'filter_nodes': {'node2'}}, 'node2'),
+    (['pathway-organizer', '--tln=node1,node2', '--fn=node3'], {'top_level_nodes': {'node1', 'node2'}, 'filter_nodes': {'node3'}}, None),
+    (['pathway-organizer', '--tln=-'], {'top_level_nodes': {'node1', 'node2', 'node3'}, 'filter_nodes': None}, 'node1\nnode2\nnode3'),
+    (['pathway-organizer', '--fn=-'], {'top_level_nodes': None, 'filter_nodes': {'node1', 'node2', 'node3'}}, 'node1\nnode2\nnode3'),
+    (['pathway-organizer', '--tln=node1,node2,node3'], {'top_level_nodes': {'node1', 'node2', 'node3'}, 'filter_nodes': None}, None),
+    (['pathway-organizer', '--fn=node1,node2,node3'], {'top_level_nodes': None, 'filter_nodes': {'node1', 'node2', 'node3'}}, None),
+    (['pathway-organizer'], {'top_level_nodes': None, 'filter_nodes': None}, None)]
+
+
+@pt.mark.parametrize('args,kwargs,stdin_mock', test_data)
+def test_print(mocker, args: list, kwargs: dict, stdin_mock: str):
+    pathway_org_mock, expected_output = _get_mock_pathway_org_and_expected_output(mocker=mocker)
+    u.test_print(
+        mocker=mocker, argv_mock=args, stdin_mock=stdin_mock, method=method, method_return_value=pathway_org_mock, method_kwargs=kwargs,
+        module=po_cli, expected_output=expected_output)
+
+
+def _get_mock_pathway_org_and_expected_output(mocker):
+    u.mock_non_instantiable(mocker=mocker)
+    hierarchy_nodes_mock: po.HierarchyNodes = {'a': {'name': 'b', 'level': 1, 'parent': 'c', 'children': ['a'], 'entry_id': 'd'}}
+    pathway_org_mock = po.PathwayOrganizer()
+    pathway_org_mock.hierarchy_nodes = hierarchy_nodes_mock
+    expected_output: str = json.dumps(hierarchy_nodes_mock, indent=2)
+    return pathway_org_mock, expected_output
+
+
+@pt.mark.parametrize('args,kwargs,stdin_mock', test_data)
+def test_file(mocker, args: list, kwargs: dict, stdin_mock: str, output_file: str):
+    pathway_org_mock, expected_output = _get_mock_pathway_org_and_expected_output(mocker=mocker)
+    u.test_file(
+        mocker=mocker, argv_mock=args, output_file=output_file, stdin_mock=stdin_mock, method=method,
+        method_return_value=pathway_org_mock, method_kwargs=kwargs, module=po_cli, expected_output=expected_output)
+
+
+@pt.mark.parametrize('args,kwargs,stdin_mock', test_data)
+def test_zip_archive(mocker, args: list, kwargs: dict, stdin_mock: str, zip_archive_data: tuple):
+    pathway_org_mock, expected_output = _get_mock_pathway_org_and_expected_output(mocker=mocker)
+    u.test_zip_archive(
+        mocker=mocker, argv_mock=args, zip_archive_data=zip_archive_data, stdin_mock=stdin_mock, method=method,
+        method_return_value=pathway_org_mock, method_kwargs=kwargs, module=po_cli, expected_output=expected_output)

dev/test_utils.py

@@ -5,6 +5,7 @@
 import dev.utils as u
 import kegg_pull.pull as p
 import kegg_pull.rest as r
+import kegg_pull.pathway_organizer as po
 
 
 @pt.mark.parametrize('comma_separated_list', [',,', ',', ''])
@@ -33,7 +34,7 @@ def test_get_range_values_exception():
 
 
 @pt.mark.parametrize(
-    'NonInstantiable,kwargs', [(p.PullResult, {}), (r.KEGGresponse, {'status': None, 'kegg_url': None})])
+    'NonInstantiable,kwargs', [(p.PullResult, {}), (r.KEGGresponse, {'status': None, 'kegg_url': None}), (po.PathwayOrganizer, {})])
 def test_non_instantiable(NonInstantiable: type, kwargs: dict):
     expected_error_message = f'The class "{NonInstantiable.__name__}" cannot be instantiated outside of its module.'
     with pt.raises(RuntimeError) as error:

docs/api.rst

@@ -19,6 +19,10 @@ API
     :members:
     :undoc-members:
 
+.. automodule:: kegg_pull.pathway_organizer
+    :members:
+    :undoc-members:
+
 .. automodule:: kegg_pull.rest
     :members:
     :undoc-members:

docs/cli.rst

@@ -79,6 +79,15 @@ If the ``--unsuccessful-threshold`` option is set and surpassed, an ``aborted-pu
     :end-before: """
     :language: none
 
+.. include:: ../src/kegg_pull/pathway_organizer.py
+    :start-after: """
+    :end-before: """
+
+.. literalinclude:: ../src/kegg_pull/pathway_organizer_cli.py
+    :start-at: Usage:
+    :end-before: """
+    :language: none
+
 .. include:: ../src/kegg_pull/rest.py
     :start-after: """
     :end-before: """

docs/tutorial.rst

@@ -320,6 +320,24 @@ operation or “conv” operation into dictionaries usable in python code.
     {'cpd:C00001': {'pubchem:3303'}, 'cpd:C00002': {'pubchem:3304'}}
 
 
+Pathway Organizer
+~~~~~~~~~~~~~~~~~
+
+The ``pathway_organizer`` module flattens a brite hierarchy into a
+mapping of the IDs of its nodes to information about those nodes.
+
+.. code:: python3
+
+    import kegg_pull.pathway_organizer as po
+    pathway_org = po.PathwayOrganizer.load_from_kegg()
+    print(pathway_org.hierarchy_nodes['Metabolism'])
+
+
+.. parsed-literal::
+
+    {'name': 'Metabolism', 'level': 1, 'parent': None, 'children': ['Amino acid metabolism', 'Biosynthesis of other secondary metabolites', 'Carbohydrate metabolism', 'Chemical structure transformation maps', 'Energy metabolism', 'Global and overview maps', 'Glycan biosynthesis and metabolism', 'Lipid metabolism', 'Metabolism of cofactors and vitamins', 'Metabolism of other amino acids', 'Metabolism of terpenoids and polyketides', 'Nucleotide metabolism', 'Xenobiotics biodegradation and metabolism'], 'entry_id': None}
+
+
 Rest API
 ~~~~~~~~
 
@@ -690,6 +708,29 @@ map
         "cpd:C00084",
 
 
+pathway-organizer
+~~~~~~~~~~~~~~~~~
+
+.. code:: none
+
+    % kegg_pull pathway-organizer --tln=Metabolism --fn="Global and overview maps,Carbohydrate metabolism" --output=hierarchy-nodes.json
+    % head hierarchy-nodes.json
+
+
+.. parsed-literal::
+
+    {
+      "path:map00190": {
+        "name": "00190  Oxidative phosphorylation",
+        "level": 3,
+        "parent": "Energy metabolism",
+        "children": null,
+        "entry_id": "path:map00190"
+      },
+      "path:map00195": {
+        "name": "00195  Photosynthesis",
+
+
 rest
 ~~~~
 

src/kegg_pull/__init__.py

@@ -7,6 +7,8 @@
 
 ``map``
 
+``pathway_organizer``
+
 ``rest``
 
 ``kegg_url``

src/kegg_pull/__main__.py

@@ -6,13 +6,15 @@
     kegg_pull pull ...              Pull, separate, and store an arbitrary number of KEGG entries to the local file system.
     kegg_pull entry-ids ...         Obtain a list of KEGG entry IDs.
     kegg_pull map ...               Obtain a mapping of entry IDs (KEGG or outside databases) to the IDs of related entries.
+    kegg_pull pathway-organizer ... Creates a flattened version of a pathways Brite hierarchy.
     kegg_pull rest ...              Executes one of the KEGG REST API operations.
 """
 import sys
 from . import __version__
 from . import pull_cli as p_cli
 from . import entry_ids_cli as ei_cli
 from . import map_cli as map_cli
+from . import pathway_organizer_cli as po_cli
 from . import rest_cli as r_cli
 
 
@@ -24,6 +26,8 @@ def main() -> None:
         ei_cli.main()
     elif first_arg == 'map':
         map_cli.main()
+    elif first_arg == 'pathway-organizer':
+        po_cli.main()
     elif first_arg == 'rest':
         r_cli.main()
     elif first_arg == '--full-help':
@@ -36,6 +40,8 @@ def main() -> None:
         print(separator)
         print(map_cli.__doc__)
         print(separator)
+        print(po_cli.__doc__)
+        print(separator)
         print(r_cli.__doc__)
     elif first_arg == '--version' or first_arg == '-v':
         print(__version__)