temporarily remove different SchemaFeatureSupport levels for simplicity. Also introduce caching of attempted sub-paths

Author: Logende

.gitignore

@@ -23,32 +23,6 @@
 from linkml.generators.sqlalchemygen import SQLAlchemyGenerator
 
 
-class ConverterLinkMlToJsonSchema(ConverterInternal):
-    def __init__(self):
-        super().__init__(
-            name="LinkML JsonSchemaGenerator",
-            service_address="internal",
-            service_name="FlaskApp",
-            source_format=SchemaLanguage.LinkMl,
-            target_format=SchemaLanguage.JsonSchema,
-            supported_features=None
-        )
-
-    def converter_logic(self, schema: str) -> str:
-        with tempfile.NamedTemporaryFile("w+", suffix=".yaml") as f:
-            f.write(schema)
-            f.flush()
-            json_schema = JsonSchemaGenerator(f.name).serialize()
-        return json_schema
-
-    def validate_input(self, schema: str) -> bool:
-        # Implement validation logic for LinkML schema
-        return True
-
-    def validate_output(self, schema: str) -> bool:
-        # Implement validation logic for JSON Schema
-        return True
-
 class ConverterFromLinkMl(ConverterInternal):
     def __init__(self, target_format: SchemaLanguage):
         super().__init__(
@@ -57,7 +31,7 @@ def __init__(self, target_format: SchemaLanguage):
             service_name="FlaskApp",
             source_format=SchemaLanguage.LinkMl,
             target_format=target_format,
-            supported_features=None
+            supported_features=set()
         )
 
     def converter_logic(self, schema: str) -> str:
@@ -102,6 +76,7 @@ def validate_output(self, schema: str) -> bool:
         # Implement validation logic for LinkML schema
         return True
 
+
 class ConverterJsonSchemaToLinkMl(ConverterInternal):
     def __init__(self):
         super().__init__(
@@ -110,7 +85,7 @@ def __init__(self):
             service_name="FlaskApp",
             source_format=SchemaLanguage.JsonSchema,
             target_format=SchemaLanguage.LinkMl,
-            supported_features=None
+            supported_features=set()
         )
 
     def converter_logic(self, schema: str) -> str:
@@ -150,7 +125,7 @@ def __init__(self):
             service_name="FlaskApp",
             source_format=SchemaLanguage.Owl,
             target_format=SchemaLanguage.LinkMl,
-            supported_features=None
+            supported_features=set()
         )
 
     def converter_logic(self, schema: str) -> str:
@@ -181,4 +156,4 @@ def validate_input(self, schema: str) -> bool:
 
     def validate_output(self, schema: str) -> bool:
         # Implement validation logic for LinkML schema
-        return True
+        return True

schema-conversion-orchestrator/ConvertersLinkMl.py

@@ -14,7 +14,7 @@ def __init__(self, target_format: SchemaLanguage):
             service_name="FlaskApp",
             source_format=SchemaLanguage.MdModels,
             target_format=target_format,
-            supported_features=None
+            supported_features=set()
         )
 
     def converter_logic(self, schema: str) -> str:
@@ -64,7 +64,7 @@ def __init__(self):
             service_name="FlaskApp",
             source_format=SchemaLanguage.JsonSchema,
             target_format=SchemaLanguage.MdModels,
-            supported_features=None
+            supported_features=set()
         )
 
     def converter_logic(self, schema: str) -> str:

schema-conversion-orchestrator/ConvertersMdModels.py

@@ -13,7 +13,7 @@ def __init__(self):
             service_name="FlaskApp",
             source_format=SchemaLanguage.Xsd,
             target_format=SchemaLanguage.JsonSchema,
-            supported_features=None
+            supported_features=set()
         )
 
     def converter_logic(self, schema: str) -> str:

schema-conversion-orchestrator/ConvertersXsd.py

@@ -2,10 +2,10 @@
 from typing import List
 from strenum import StrEnum
 
-from converter import (ConversionResult, ConversionResults, ConversionPaths, Converter)
+from converter import (ConversionResult, ConversionResults, ConversionPaths, Converter, conversion_path_to_string, ConversionsCache)
 from schema_types import SchemaLanguage
 from logic import identify_schema_features, rank_paths
-from app import DETAILED_ERROR_OUTPUT, schema_languages_features
+from app import DETAILED_ERROR_OUTPUT
 
 
 class ConversionStrategy(StrEnum):
@@ -20,8 +20,9 @@ def convert_with_strategy_most_features_preserved(source: SchemaLanguage, target
     doc_features = set(identify_schema_features(schema, source))
     if not doc_features:
         print("Warning: No schema feature identification available for the given schema format '" + source + "'.")
-    ranked_paths = rank_paths(paths, doc_features, schema_languages_features)
+    ranked_paths = rank_paths(paths, doc_features)
     all_attempts: List[ConversionResult] = []
+    conversions_cache = {}  # cache for all conversion sub-paths
     result_schema = None
 
     # attempt conversion via best path and if it fails, try remaining paths and print error message only to console
@@ -30,7 +31,9 @@ def convert_with_strategy_most_features_preserved(source: SchemaLanguage, target
     while result_schema is None and len(ranked_paths) > 0:
         best_path, unsupported_features = ranked_paths[0]
         try:
-            result_schema = attempt_conversion_path(source, target, best_path, schema)
+            result_schema, conversions_cache_update = attempt_conversion_path(source, target, best_path, schema,
+                                                                              conversions_cache)
+            conversions_cache = conversions_cache_update
 
             if not result_schema:
                 all_attempts.append((False, "Conversion resulted in 'None' schema.", best_path))
@@ -56,10 +59,13 @@ def convert_with_strategy_least_character_loss(source: SchemaLanguage, target: S
     Does not stop at success but explores all paths. Trivial feature loss strategy which is character based.
     Much less effort than a proper feature loss analysis and still effective."""
     all_attempts: List[ConversionResult] = []
+    conversions_cache = {}  # cache for all conversion sub-paths
     for path in paths:
         result_schema = None
         try:
-            result_schema = attempt_conversion_path(source, target, path, schema)
+            result_schema, conversions_cache_update = attempt_conversion_path(source, target, path, schema,
+                                                                              conversions_cache)
+            conversions_cache = conversions_cache_update
 
             if not result_schema:
                 all_attempts.append((False, "Conversion resulted in 'None' schema.", path))
@@ -78,17 +84,41 @@ def convert_with_strategy_least_character_loss(source: SchemaLanguage, target: S
     return all_attempts
 
 
-def attempt_conversion_path(source: str, target: str, path: List[Converter], schema: str) -> str:
+# the conversions cache contains the results of all previously attempted conversion sub-paths
+def attempt_conversion_path(source: str, target: str, path: List[Converter], schema: str,
+                            conversions_cache: ConversionsCache) -> tuple[str, ConversionsCache]:
     print_conversion_path(source, target, path)
     current_schema = schema
     current_converter = None
+    conversion_sub_path = []
     try:
         for conv in path:
             current_converter = conv
-            current_schema = conv.convert(current_schema)
-            print(
-                "Intermediate schema of format " + conv.target_format + " after conversion via " + conv.service_name + ": " + current_schema)
-        return current_schema
+            conversion_sub_path.append(conv)
+
+            # check cache
+            conversion_sub_path_hash = conversion_path_to_string(conversion_sub_path)
+            if conversion_sub_path_hash in conversions_cache:
+                # cache hit
+                cached_result = conversions_cache[conversion_sub_path_hash]
+                if cached_result is None:
+                    # previously failed conversion for this sub-path
+                    raise Exception("Previously failed conversion for this sub-path.")
+                else:
+                    # use cached result in case of cache hit and good previous conversion
+                    current_schema = cached_result
+                    print(
+                        "Using cached intermediate schema of format " + conv.target_format + " after conversion via " + conv.service_name + ": " + current_schema)
+                    continue
+            else:
+                # cache miss - perform conversion
+                current_schema = conv.convert(current_schema)
+                print(
+                    "Intermediate schema of format " + conv.target_format + " after conversion via " + conv.service_name + ": " + current_schema)
+                # store in cache
+                conversions_cache[conversion_sub_path_hash] = current_schema
+
+        return current_schema, conversions_cache
     except Exception as e:
         print(
             "Conversion failed at step from " + current_converter.source_format + " to " + current_converter.target_format + " via " + current_converter.service_name + " because of error: " + str(

schema-conversion-orchestrator/conversion_strategies.py

@@ -2,9 +2,8 @@
 import os
 import subprocess
 import tempfile
-from typing import List, Dict, Tuple
-from schema_types import SchemaLanguage, SchemaFeature, SchemaFeatureSupport
-
+from typing import List, Tuple
+from schema_types import SchemaLanguage, SchemaFeature
 
 
 class Converter:
@@ -19,7 +18,7 @@ class Converter:
     """
 
     def __init__(self, name: str, service_address: str, service_name: str, source_format: SchemaLanguage,
-                 target_format: SchemaLanguage, supported_features: Dict[SchemaFeature, SchemaFeatureSupport] | None):
+                 target_format: SchemaLanguage, supported_features: set[SchemaFeature]):
         self.name = name
         self.service_address = service_address
         self.service_name = service_name
@@ -33,7 +32,7 @@ def convert(self, schema: str) -> str:
 
 class ConverterExternal(Converter):
     def __init__(self, name: str, executable_path: str, service_name: str, source_format: SchemaLanguage,
-                 target_format: SchemaLanguage, supported_features: Dict[SchemaFeature, SchemaFeatureSupport] | None):
+                 target_format: SchemaLanguage, supported_features: set[SchemaFeature]):
         super().__init__(name, executable_path, service_name, source_format, target_format, supported_features)
         self.executable_path = executable_path
 
@@ -80,7 +79,7 @@ def convert(self, schema: str) -> str:
 
 class ConverterInternal(Converter):
     def __init__(self, name: str, service_address: str, service_name: str, source_format: SchemaLanguage,
-                 target_format: SchemaLanguage, supported_features: Dict[SchemaFeature, SchemaFeatureSupport] | None):
+                 target_format: SchemaLanguage, supported_features: set[SchemaFeature]):
         super().__init__(name, service_address, service_name, source_format, target_format, supported_features)
 
     def convert(self, schema: str) -> str:
@@ -111,7 +110,7 @@ class ConverterExternalGeneric(ConverterExternal):
     """Generic external converter that can handle multiple conversion types"""
 
     def __init__(self, name: str, executable_path: str, source_format: SchemaLanguage,
-                 target_format: SchemaLanguage, supported_features: Dict[SchemaFeature, SchemaFeatureSupport] | None,
+                 target_format: SchemaLanguage, supported_features: set[SchemaFeature],
                  converter_type: str):
         super().__init__(name, executable_path, converter_type, source_format, target_format, supported_features)
         self.converter_type = converter_type
@@ -177,3 +176,9 @@ def convert(self, schema: str) -> str:
 
 # List of results ranked by success
 ConversionResults = List[ConversionResult]
+
+ConversionsCache = dict[str, str | None]
+
+
+def conversion_path_to_string(path: ConversionPath) -> str:
+    return " -> ".join([f"{conv.source_format.value} to {conv.target_format.value} via {conv.service_name}" for conv in path])

schema-conversion-orchestrator/converter.py

@@ -1,6 +1,6 @@
 from typing import List, Set, Tuple, Dict
 from converter import Converter, ConversionGraph, ConversionPath, ConversionPaths
-from schema_types import SchemaLanguage, SchemaFeature, SchemaLanguagesFeatures, SchemaFeatureSupport
+from schema_types import SchemaLanguage, SchemaFeature
 from identify_schema_features_json_schema import identify_schema_features_json_schema
 
 
@@ -56,13 +56,8 @@ def find_paths(source: SchemaLanguage, target: SchemaLanguage, conversion_graph:
 
 # ranks the paths based on how many features they support from the document schema plus returns a list of unsupported
 # features for each path
-# Feature support is not a yes or no but an enum of preserved, approximated, weakened, lost.
-# We start with the state of features of the user document. Then for each path step we remove the unsupported features and apply
-# downgrades to the others, depending on 1. feature support by the schema language of this step and 2. feature support
-# by the conversion step itself.
-def rank_paths(paths: ConversionPaths, doc_features: set[SchemaFeature] | None,
-               schema_languages_features: SchemaLanguagesFeatures) -> List[
-    Tuple[ConversionPath, Dict[SchemaFeature, SchemaFeatureSupport]]]:
+def rank_paths(paths: ConversionPaths, doc_features: set[SchemaFeature] | None) -> List[
+    Tuple[ConversionPath, set[SchemaFeature]]]:
     if doc_features is None:
         # todo: assume ALL features in case of no known list of features
         doc_features = set()

schema-conversion-orchestrator/logic.py

@@ -53,10 +53,10 @@ def register_external_converters() -> List[Converter]:
         for info in converter_infos:
             try:
                 # Parse supported features
-                features = {}
-                for feature_str in info.get('supportedFeatures', {}):
+                features = set()
+                for feature_str in info.get('supportedFeatures', []):
                     try:
-                        features[SchemaFeature(feature_str)] = info['supportedFeatures'][feature_str]
+                        features.add(SchemaFeature(feature_str))
                     except ValueError:
                         print(f"Unknown feature: {feature_str}")
 
@@ -85,10 +85,10 @@ def register_external_converters() -> List[Converter]:
         for info in converter_infos:
             try:
                 # Parse supported features
-                features = {}
-                for feature_str in info.get('supportedFeatures', {}):
+                features = set()
+                for feature_str in info.get('supportedFeatures', []):
                     try:
-                        features[SchemaFeature(feature_str)] = info['supportedFeatures'][feature_str]
+                        features.add(SchemaFeature(feature_str))
                     except ValueError:
                         print(f"Unknown feature: {feature_str}")
 
@@ -107,4 +107,4 @@ def register_external_converters() -> List[Converter]:
     else:
         print(f"Java converter not found at: {java_jar}")
 
-    return converters
+    return converters

schema-conversion-orchestrator/register_external_converters.py

@@ -1,5 +1,5 @@
 from typing import List
-from ConvertersLinkMl import ConverterFromLinkMl, ConverterJsonSchemaToLinkMl, ConverterLinkMlToJsonSchema, ConverterOwlToLinkMl
+from ConvertersLinkMl import ConverterFromLinkMl, ConverterJsonSchemaToLinkMl, ConverterOwlToLinkMl
 from ConvertersMdModels import ConverterFromMdModels, ConverterJsonSchemaToMdModels
 
 from converter import Converter
@@ -38,6 +38,4 @@ def register_python_converters() -> List[Converter]:
 
         ConverterJsonSchemaToMdModels(),
 
-        ConverterLinkMlToJsonSchema() # redundant
-
-    ]
+    ]