different improvements and remove old code

Author: Logende

generate_diagrams.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+
+python3 schema-conversion-orchestrator/output_generate_diagrams.py

schema-conversion-orchestrator/app.py

@@ -1,6 +1,6 @@
 from conversion_strategies import ConversionStrategy, \
     convert_with_strategy_least_character_loss
-from converter import (Converter, ConverterExternal, prepare_conversion_results_for_serializing)
+from converter import (Converter, prepare_conversion_results_for_serializing)
 from schema_types import schema_language_from_string
 from logic import build_conversion_graph, find_paths
 from register_converters import register_converters
@@ -25,23 +25,6 @@ def health():
     return {"status": "ok"}, 200
 
 
-@app.route("/registerConversion", methods=["POST"])
-def register_conversion():
-    data = request.json
-    conv = ConverterExternal(
-        data["name"],
-        data["serviceAddress"],
-        data["sourceFormat"],
-        data["targetFormat"],
-    )
-    converters.append(conv)
-    global conversion_graph
-    conversion_graph = build_conversion_graph(converters)
-    print(
-        f"Registered new converter: {conv.name} from {conv.source_format} to {conv.target_format} at {conv.service_address}.")
-    return {"status": "registered"}, 200
-
-
 @app.route("/convert", methods=["POST"])
 def convert():
     data = request.json

schema-conversion-orchestrator/conversion_strategies.py

@@ -2,10 +2,12 @@
 from typing import List
 from strenum import StrEnum
 
-from converter import (ConversionResult, ConversionResults, ConversionPaths, Converter, conversion_path_to_string, ConversionsCache)
+from converter import (ConversionResult, ConversionResults, ConversionPaths, Converter, conversion_path_to_string,
+                       ConversionsCache)
 from schema_types import SchemaLanguage
 
 DETAILED_ERROR_OUTPUT = False
+DETAILED_RESULT_OUTPUT = True
 
 
 class ConversionStrategy(StrEnum):
@@ -20,7 +22,6 @@ def convert_with_strategy_least_character_loss(source: SchemaLanguage, target: S
     all_attempts: List[ConversionResult] = []
     conversions_cache = {}  # cache for all conversion sub-paths
     for path in paths:
-        result_schema = None
         try:
             result_schema, conversions_cache_update = attempt_conversion_path(source, target, path, schema,
                                                                               conversions_cache)
@@ -40,6 +41,18 @@ def convert_with_strategy_least_character_loss(source: SchemaLanguage, target: S
     # sort all attempts by success: success first. Then by length of resulting schema (descending)
     all_attempts.sort(key=lambda x: (not x[0], -len(x[1]) if x[0] else float('inf')))
 
+    # print overall result: how many succeeded/failed and their character lengths
+    success_count = sum(1 for attempt in all_attempts if attempt[0])
+    failure_count = len(all_attempts) - success_count
+    print(f"Conversion attempts completed: {success_count} succeeded, {failure_count} failed.")
+    for i, attempt in enumerate(all_attempts):
+        success, result_schema_or_error, path = attempt
+        if success:
+            print(
+                f"- Attempt {i + 1} ({conversion_path_to_string(path)}): Success, Resulting schema length: {len(result_schema_or_error)} characters.")
+        else:
+            print(f"- Attempt {i + 1} ({conversion_path_to_string(path)}): Failure, Error: {result_schema_or_error}")
+
     return all_attempts
 
 
@@ -72,8 +85,9 @@ def attempt_conversion_path(source: str, target: str, path: List[Converter], sch
             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)
+                if DETAILED_RESULT_OUTPUT:
+                    print(
+                        "\n\nIntermediate schema of format " + conv.target_format + " after conversion via " + conv.service_name + ": \n" + current_schema + "\n\n\n\n")
                 # store in cache
                 conversions_cache[conversion_sub_path_hash] = current_schema
 
@@ -88,6 +102,6 @@ def attempt_conversion_path(source: str, target: str, path: List[Converter], sch
 
 
 def print_conversion_path(source: str, target: str, path: List[Converter]) -> None:
-    print("Given the source format " + source + " and target format " + target + ", the best available path is:")
+    print("Conversion path for source format '" + source + "' and target format '" + target + "':")
     for conv in path:
-        print(f"{conv.source_format} -> {conv.target_format} via {conv.name} ({conv.service_address})")
+        print(f"- {conv.source_format} --({conv.name} ({conv.service_address})--> {conv.target_format})")

schema-conversion-orchestrator/converter.py

@@ -30,6 +30,14 @@ def convert(self, schema: str) -> str:
 
 
 class ConverterExternal(Converter):
+    """
+    External converter that calls a subprocess to perform conversion
+    :param name: Name of the converter
+    :param executable_path: Path to the converter executable
+    :param service_name: Name of the converter service
+    :param source_format: Source schema language
+    :param target_format: Target schema language
+    """
     def __init__(self, name: str, executable_path: str, service_name: str, source_format: SchemaLanguage,
                  target_format: SchemaLanguage):
         super().__init__(name, executable_path, service_name, source_format, target_format)
@@ -77,6 +85,14 @@ def convert(self, schema: str) -> str:
 
 
 class ConverterInternal(Converter):
+    """
+    Internal converter that performs conversion using built-in logic, in Python
+    :param name: Name of the converter
+    :param service_address: Address of the converter service or path to executable
+    :param service_name: Name of the converter service
+    :param source_format: Source schema language
+    :param target_format: Target schema language
+    """
     def __init__(self, name: str, service_address: str, service_name: str, source_format: SchemaLanguage,
                  target_format: SchemaLanguage):
         super().__init__(name, service_address, service_name, source_format, target_format)

…-orchestrator/build_conversion_matrix.py …trator/output_build_conversion_matrix.pyschema-conversion-orchestrator/build_conversion_matrix.py renamed to schema-conversion-orchestrator/output_build_conversion_matrix.py schema-conversion-orchestrator/build_conversion_matrix.py renamed to schema-conversion-orchestrator/output_build_conversion_matrix.py

@@ -39,7 +39,7 @@ def build_conversion_matrix(conversion_graph: ConversionGraph) -> pd.DataFrame:
     return matrix
 
 
-def plot_conversion_matrix(matrix: pd.DataFrame):
+def plot_conversion_matrix(matrix: pd.DataFrame, output_path: str = None) -> None:
     # Build weighted numeric matrix for coloring
     numeric_matrix = matrix.copy()
 
@@ -76,20 +76,8 @@ def plot_conversion_matrix(matrix: pd.DataFrame):
     plt.xticks(rotation=45, ha="right")
     plt.yticks(rotation=0)
     plt.tight_layout()
-    plt.show()
-
-
-if __name__ == "__main__":
-    from register_converters import register_converters
-    from logic import build_conversion_graph
-
-    converters: List[Converter] = register_converters()
-    conversion_graph: ConversionGraph = build_conversion_graph(converters)
-
-    df = build_conversion_matrix(conversion_graph)
-
-    # Print in console
-    print(df.to_string())
-
-    # Plot nicely
-    plot_conversion_matrix(df)
+    if output_path:
+        plt.savefig(output_path)
+    else:
+        plt.show()
+    plt.close()

schema-conversion-orchestrator/output_conversion_graph_to_rdf.py

@@ -0,0 +1,46 @@
+from rdflib import Graph, Namespace, Literal
+from rdflib.namespace import RDF, RDFS
+from converter import ConversionGraph
+
+
+def conversion_graph_to_rdf(conversion_graph: ConversionGraph, output_path):
+    EX = Namespace("http://example.org/conversions/")
+
+    g = Graph()
+    g.bind("ex", EX)
+    g.bind("rdfs", RDFS)
+
+    # define a class and a property for conversions
+    CONVERSION = EX.Conversion
+    HAS_NAME = EX.name
+    HAS_SOURCE = EX.sourceFormat
+    HAS_TARGET = EX.targetFormat
+
+    for source, converters in conversion_graph.items():
+        for converter in converters:
+            src = string_normalize(str(converter.source_format))
+            tgt = string_normalize(str(converter.target_format))
+
+            # create URIs for formats
+            src_uri = EX[f"format/{src}"]
+            tgt_uri = EX[f"format/{tgt}"]
+
+            # add label for readability
+            g.add((src_uri, RDFS.label, Literal(src)))
+            g.add((tgt_uri, RDFS.label, Literal(tgt)))
+
+            # create a conversion node (reified edge)
+            conv_uri = EX[f"conversion/{string_normalize(converter.name)}"]
+
+            g.add((conv_uri, RDF.type, CONVERSION))
+            g.add((conv_uri, HAS_NAME, Literal(converter.name)))
+            g.add((conv_uri, HAS_SOURCE, src_uri))
+            g.add((conv_uri, HAS_TARGET, tgt_uri))
+
+    # serialize as Turtle to file
+    g.serialize(output_path, format="turtle")
+
+
+# replace intermediate whitespaces by underline and trim and avoid other issues to make it a valid URI fragment
+def string_normalize(s: str) -> str:
+    return "_".join(s.strip().split())

schema-conversion-orchestrator/output_generate_diagrams.py

@@ -0,0 +1,42 @@
+import os
+from typing import List
+from converter import ConversionGraph, Converter
+from output_visualize_conversion_graph import visualize_conversion_graph
+from output_conversion_graph_to_rdf import conversion_graph_to_rdf
+from output_build_conversion_matrix import build_conversion_matrix, plot_conversion_matrix
+from register_converters import register_converters
+from logic import build_conversion_graph
+
+
+def generate_diagrams(conversion_graph: ConversionGraph, outputs_dir: str):
+    conversion_graph_img_output_path = os.path.join(outputs_dir, "conversion_graph.png")
+    turtle_output_path = os.path.join(outputs_dir, "conversion_graph.ttl")
+    conversion_matrix_path = os.path.join(outputs_dir, "conversion_matrix.png")
+
+    visualize_conversion_graph(conversion_graph, output_path=conversion_graph_img_output_path)
+    conversion_graph_to_rdf(conversion_graph, output_path=turtle_output_path)
+
+    conversion_matrix = build_conversion_matrix(conversion_graph)
+    plot_conversion_matrix(conversion_matrix, output_path=conversion_matrix_path)
+
+
+
+if __name__ == "__main__":
+    converters_full: List[Converter] = register_converters()
+    conversion_graph_full: ConversionGraph = build_conversion_graph(converters_full)
+
+    converters_core: List[Converter] = register_converters(True)
+    conversion_graph_core: ConversionGraph = build_conversion_graph(converters_core)
+
+    # dir of current file
+    python_code_dir = os.path.dirname(os.path.abspath(__file__))
+    root_dir = os.path.dirname(python_code_dir)
+    outputs_dir_full = os.path.join(root_dir, "outputs", "full")
+    outputs_dir_core = os.path.join(root_dir, "outputs", "core")
+
+    # create output directories if they don't exist
+    os.makedirs(outputs_dir_full, exist_ok=True)
+    os.makedirs(outputs_dir_core, exist_ok=True)
+
+    generate_diagrams(conversion_graph_full, outputs_dir_full)
+    generate_diagrams(conversion_graph_core, outputs_dir_core)

…chestrator/visualize_conversion_graph.py …tor/output_visualize_conversion_graph.pyschema-conversion-orchestrator/visualize_conversion_graph.py renamed to schema-conversion-orchestrator/output_visualize_conversion_graph.py schema-conversion-orchestrator/visualize_conversion_graph.py renamed to schema-conversion-orchestrator/output_visualize_conversion_graph.py

@@ -2,11 +2,10 @@
 matplotlib.use("TkAgg")  # or "Qt5Agg"
 import matplotlib.pyplot as plt
 import networkx as nx
-from typing import List
-from converter import Converter, ConversionGraph
+from converter import ConversionGraph
 
 
-def visualize_conversion_graph(conversion_graph: ConversionGraph):
+def visualize_conversion_graph(conversion_graph: ConversionGraph, output_path):
     # Build directed graph
     G = nx.DiGraph()
     edge_labels = {}
@@ -73,12 +72,7 @@ def visualize_conversion_graph(conversion_graph: ConversionGraph):
     plt.title("Schema Conversion Graph", fontsize=16)
     plt.axis("off")
     plt.tight_layout()
-    plt.show()
 
+    # save plot to file
+    plt.savefig(output_path)
 
-if __name__ == "__main__":
-    from register_converters import register_converters
-    from logic import build_conversion_graph
-    converters: List[Converter] = register_converters()
-    conversion_graph: ConversionGraph = build_conversion_graph(converters)
-    visualize_conversion_graph(conversion_graph)

schema-conversion-orchestrator/register_converters.py

@@ -1,10 +1,30 @@
 from typing import List
 from converter import Converter
+from schema_types import SchemaLanguage
 from register_python_converters import register_python_converters
 from register_external_converters import register_external_converters
 
 
-def register_converters():
+CORE_SCHEMA_LANGUAGES: List[SchemaLanguage] = [
+    SchemaLanguage.JsonSchema,
+    SchemaLanguage.Owl,
+    SchemaLanguage.Xsd,
+    SchemaLanguage.SHACL,
+    SchemaLanguage.MdModels,
+    SchemaLanguage.LinkMl,
+    SchemaLanguage.JsonLD
+]
+
+
+def register_converters(only_core_languages: bool = False) -> List[Converter]:
     converters: List[Converter] = register_python_converters()
     converters.extend(register_external_converters())
+
+    # remove converters that do involve non-core schema languages if only_core_languages is True
+    if only_core_languages:
+        converters = [
+            conv for conv in converters
+            if conv.source_format in CORE_SCHEMA_LANGUAGES and conv.target_format in CORE_SCHEMA_LANGUAGES
+        ]
+
     return converters