Compact XSD strings

case_example_mapping/__init__.py

@@ -1 +1,5 @@
 __version__ = "0.0.1"
+
+from typing import Union
+
+JSON = Union[None, bool, dict[str, "JSON"], float, int, list["JSON"], str]

case_example_mapping/rdf_to_case.py

@@ -1,8 +1,69 @@
+import json
 import sys
+from io import TextIOWrapper
 
 from rdflib import Graph, Literal, Namespace, URIRef
 from rdflib.namespace import RDF, XSD
 
+from case_example_mapping import JSON
+
+STR_XSD_STRING = str(XSD.string)
+
+
+def compact_xsd_strings(json_object: JSON) -> JSON:
+    """
+    Some JSON-LD serializers export all literals with datatype
+    annotations.  A string-literal without a datatype can be optionally
+    exported with a datatype annotation of ``xsd:string``, or left to be
+    interpreted as a ``xsd:string`` by default.  This function compacts
+    string-literals that are ``xsd:string``-typed into a JSON String
+    instead of a JSON Object.
+
+    >>> from rdflib import XSD
+    >>> x = {
+    ...     "@id": "http://example.org/s",
+    ...     "http://example.org/p": [
+    ...         {
+    ...             "@type": str(XSD.string),
+    ...             "@value": "o"
+    ...         }
+    ...     ]
+    ... }
+    >>> x
+    {'@id': 'http://example.org/s', 'http://example.org/p': [{'@type': 'http://www.w3.org/2001/XMLSchema#string', '@value': 'o'}]}
+    >>> compact_xsd_strings(x)
+    {'@id': 'http://example.org/s', 'http://example.org/p': ['o']}
+    """
+    if json_object is None:
+        return json_object
+    elif isinstance(json_object, (bool, float, int, str)):
+        return json_object
+    elif isinstance(json_object, dict):
+        if "@value" in json_object:
+            # Reviewing Literal.
+            if json_object.get("@type") == STR_XSD_STRING:
+                assert isinstance(json_object["@value"], str)
+                return json_object["@value"]
+            else:
+                return json_object
+        else:
+            return {k: compact_xsd_strings(json_object[k]) for k in json_object.keys()}
+    elif isinstance(json_object, list):
+        return [compact_xsd_strings(x) for x in json_object]
+    else:
+        raise TypeError("Unexpected type of object: %s." % type(json_object))
+
+
+def serialize_jsonld(graph: Graph, fh: TextIOWrapper) -> None:
+    """
+    This function serializes the graph with a step taken to compact
+    ``xsd:string`` Literals that became JSON Objects.
+    """
+    json_string = graph.serialize(format="json-ld")
+    json_object = json.loads(json_string)
+    json_object = compact_xsd_strings(json_object)
+    json.dump(json_object, fh, indent=4)
+
 
 def main() -> None:
     # Process the command line arguments to get the output path
@@ -52,7 +113,7 @@ def main() -> None:
     # Write the CASE graph to a file
     try:
         with open(output_path, "w") as case_file:
-            case_file.write(g.serialize(format="json-ld", indent=4))
+            serialize_jsonld(g, case_file)
             print(f"CASE graph exported to: {output_path}")
     except IOError:
         print(f"Error writing to path: {output_path}")