Assign edge knowledge sources with more automation

kgx/source/json_source.py

@@ -1,19 +1,9 @@
 import gzip
 import typing
-from typing import Optional, Generator, Any
 import ijson
 from itertools import chain
-from typing import Dict, Tuple, Any, Generator, Optional, List
+from typing import Any, Generator, Optional
 from kgx.config import get_logger
-from kgx.error_detection import ErrorType, MessageLevel
-from kgx.source.source import Source
-from kgx.utils.kgx_utils import (
-    generate_uuid,
-    generate_edge_key,
-    extension_types,
-    archive_read_mode,
-    sanitize_import
-)
 from kgx.source.tsv_source import TsvSource
 log = get_logger()
 

kgx/transformer.py

@@ -3,9 +3,10 @@
 from os.path import exists
 from sys import stderr
 from typing import Dict, Generator, List, Optional, Callable, Set
-
+from bmt import Toolkit
 from kgx.config import get_logger
 from kgx.error_detection import ErrorType, MessageLevel, ErrorDetecting
+from pprint import pprint
 from kgx.source import (
     GraphSource,
     Source,
@@ -110,6 +111,8 @@ def __init__(
         self._seen_nodes = set()
         self._infores_catalog: Dict[str, str] = dict()
 
+        self.toolkit = Toolkit()
+
         if infores_catalog and exists(infores_catalog):
             with open(infores_catalog, "r") as irc:
                 for entry in irc:
@@ -118,6 +121,11 @@ def __init__(
                         if entry:
                             source, infores = entry.split("\t")
                             self._infores_catalog[source] = infores
+            for k, v in self.toolkit.infores_map.items():
+                self._infores_catalog[k] = v.get("name").strip()
+        else:
+            for k, v in self.toolkit.infores_map.items():
+                self._infores_catalog[k] = v.get("name").strip()
 
     def transform(
         self,

kgx/utils/infores.py

@@ -5,7 +5,6 @@
 from typing import Optional, Tuple, Callable, Dict, List, Any
 
 from kgx.utils.kgx_utils import knowledge_provenance_properties, column_types
-from kgx.error_detection import ErrorType, MessageLevel
 from kgx.config import get_logger
 
 log = get_logger()
@@ -134,7 +133,7 @@ def _get_infores(source: str) -> str:
 
             def _process_infores(source: str) -> str:
                 """
-                Process a single knowledge Source name  string into an infores, by applying rules
+                Process a single knowledge Source name string into an infores, by applying rules
                 in the _infores_processor() closure context, followed by standard formatting.
 
                 Parameters
@@ -430,8 +429,9 @@ def set_node_provenance(self, node_data: Dict):
 
     def set_edge_provenance(self, edge_data: Dict):
         """
-        Sets the edge knowledge_source value for the current edge. Edge knowledge_source properties
-        include the 'knowledge_source' related properties.
+        Sets the edge knowledge_source value for the current edge.
+        Edge knowledge_source properties include the
+        'knowledge_source' related properties.
 
         Parameters
         ----------