Refactor code (#34)

* package initialization added

* WoT toolchain CLI updated with click library

* wotis package created

* CLI interface changed to click and refractoring

* CLI interface changed to click and refractoring

* configs added to initiation

* gitattributes added

* json-ld context postprocessing refractored

* package update

.gitattributes

@@ -0,0 +1 @@
+*.html linguist-detectable=false

main.py

@@ -1,6 +1,8 @@
 import argparse
+import click
 import json
 import subprocess
+from pathlib import Path
 
 from linkml.generators.jsonschemagen import JsonSchemaGenerator
 from linkml.generators.shaclgen import ShaclGenerator
@@ -10,10 +12,7 @@
 from linkml.generators.linkmlgen import LinkmlGenerator
 from linkml_runtime.utils.schemaview import SchemaView
 from linkml_runtime.linkml_model.meta import AnonymousSlotExpression
-from pathlib import Path
-from pyld import jsonld
 
-from linkml_runtime.linkml_model.meta import SlotDefinition
 
 RESOURCES_PATH = Path('resources')
 GENS_PATH = RESOURCES_PATH / 'gens'

pyproject.toml

@@ -1,26 +1,36 @@
 [project]
-name = "thing_description_schema"
+name = "wotis"
 version = "0.1.0"
-description = "Thing Description Information Model as a LinkML schema"
+description = "A CLI for Web of Things Integrated Schemas (WOTIS)"
 authors = [
-    { name = "Mahda Noura", email = "mahdanoura@gmail.com" }
+    { name = "Mahda Noura", email = "mahda.noura@siemens.com" }
 ]
 license = "MIT"
 readme = "README.md"
-include = ["README.md", "src/thing_description_schema/schema", "project"]
-
-requires-python = ">=3.12"
+include = ["README.md", "resources/schemas/*"]
+requires-python = ">=3.11"
 dependencies = [
     "linkml-runtime>=1.8.2",
     "linkml>=1.8.2",
     "mkdocs-mermaid2-plugin>=1.1.1",
     "mkdocs-material>=9.5.32",
-    "schemasheets>=0.3.1",
+    "schemasheets>=0.3.1"
+]
+
+[project.optional-dependencies]
+cli = [
+    "click>=8.1.7"
 ]
 
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"
 
 [tool.hatch.build]
-only-include = ["main.py"]
+only-include = ["src/wotis/cli.py"]
+
+[tool.hatch.metadata]
+root = "src"
+
+[project.scripts]
+wotis = "src.wotis.cli:main"

src/wotis/__init__.py

@@ -0,0 +1,9 @@
+from pathlib import Path
+
+RESOURCES_PATH = Path('resources')
+GENS_PATH = RESOURCES_PATH / 'gens'
+SCHEMA_PATH = RESOURCES_PATH / 'schemas'
+YAML_SCHEMA_PATH = SCHEMA_PATH / 'thing_description.yaml'
+DOCDIR = GENS_PATH / 'docs' / 'ontology'
+
+GENERATORS = ['jsonschema', 'shacl', 'jsonldcontext', 'linkml']

src/wotis/cli.py

@@ -0,0 +1,131 @@
+"""Command line interface for WoTIS."""
+import click
+import logging
+from pathlib import Path
+import subprocess
+
+import yaml
+from linkml.generators.jsonschemagen import JsonSchemaGenerator
+from linkml.generators.shaclgen import ShaclGenerator
+from linkml.generators.owlgen import OwlSchemaGenerator
+from linkml.generators.jsonldcontextgen import ContextGenerator
+from linkml.generators.docgen import DocGenerator
+from linkml.generators.linkmlgen import LinkmlGenerator
+from linkml_runtime.utils.schemaview import SchemaView
+
+from src.wotis import YAML_SCHEMA_PATH, GENS_PATH, GENERATORS, DOCDIR
+from src.wotis.post_processors.jsonld_context_postprocessor import post_process_jsonld_context
+
+
+input_option = click.option('-i', '--input_schema',
+                            type=str,
+                            show_default=True,
+                            help="Path to the input schema specified as LinkML yaml.",
+                            default=YAML_SCHEMA_PATH)
+docs_option = click.option('-d',
+                           '--generate_docs',
+                           type=bool,
+                           is_flag=True,
+                           default=False,
+                           show_default=True,
+                           help="Boolean for local documentation generation.")
+serve_docs_option = click.option('-s',
+                                 '--serve_docs',
+                                 type=bool,
+                                 is_flag=True,
+                                 default=False,
+                                 show_default=True,
+                                 help="Boolean for serving the generated documentation.")
+
+
+def serve_documentation():
+    subprocess.run(['mkdocs', 'serve'], check=True)
+
+
+def generate_documentation():
+    DOCDIR.mkdir(parents=True, exist_ok=True)
+    doc_generator = DocGenerator(YAML_SCHEMA_PATH, mergeimports=False)
+    doc_generator.serialize(directory=str(DOCDIR))
+
+
+# Main generation function
+def run_generator(schema_view, generator, output_dir):
+    if generator == 'jsonschema':
+        logging.info(f"Proceeding with LinkML to JSON Schema convertion")
+        json_schema_generator = JsonSchemaGenerator(schema_view.schema, mergeimports=True)
+        (output_dir / 'jsonschema.json').write_text(json_schema_generator.serialize())
+    elif generator == 'shacl':
+        logging.info(f"Proceeding with LinkML to SHACL convertion")
+        shacl_generator = ShaclGenerator(schema_view.schema, mergeimports=False, closed=True, suffix='Shape')
+        (output_dir / 'shapes.shacl.ttl').write_text(shacl_generator.serialize())
+    elif generator == 'owl':
+        logging.info(f"Proceeding with LinkML to OWL convertion")
+        owl_generator = OwlSchemaGenerator(schema_view.schema, )
+        (output_dir / 'ontology.owl.ttl').write_text(owl_generator.serialize())
+    elif generator == 'jsonldcontext':
+        logging.info(f"Proceeding with LinkML to JSON-LD Context convertion")
+        context_generator = ContextGenerator(schema_view.schema, mergeimports=True)
+        (output_dir / 'context.jsonld').write_text(post_process_jsonld_context(schema_view,
+                                                                               context_generator.serialize()))
+    elif generator == 'linkml':
+        linkml_generator = LinkmlGenerator(schema_view.schema, mergeimports=True, format='yaml', output='linkml.yaml')
+        (output_dir / 'linkml.yaml').write_text(linkml_generator.serialize())
+    else:
+        print(f"Unknown generator: {generator}")
+
+
+@click.group()
+@click.option("-v", "--verbose", count=True)
+@click.option("-q", "--quiet")
+def main(verbose: int, quiet: bool):
+    """CLI for WOTIS (Web of Things Integrated Schemas) toolchain.
+
+    :param verbose: Verbosity while running.
+    :param quiet: Boolean to be quiet or verbose.
+    """
+    logger = logging.getLogger()
+    if verbose >= 2:
+        logger.setLevel(level=logging.DEBUG)
+    elif verbose == 1:
+        logger.setLevel(level=logging.INFO)
+    else:
+        logger.setLevel(level=logging.WARNING)
+    if quiet:
+        logger.setLevel(level=logging.ERROR)
+    logger.info(f"Logger {logger.name} set to level {logger.level}")
+
+
+@main.command()
+@input_option
+@docs_option
+@serve_docs_option
+def generate_wot_resources(input_schema: str, generate_docs: bool, serve_docs: bool):
+    """
+    Generating WoT resources (RDF, JSON-LD Context, SHACL Shapes, and JSON Schema) from manually constructed
+    LinkML-based schemas.
+    """
+    if input_schema and not Path(input_schema).exists():
+        raise FileNotFoundError(f"Cannot find input LinkML schema file {input_schema}.")
+    elif not input_schema and not YAML_SCHEMA_PATH.exists():
+        raise FileNotFoundError(f"Cannot find the default LinkML schema file {YAML_SCHEMA_PATH}.")
+    else:
+        try:
+            linkml_schema_view = SchemaView(input_schema, merge_imports=True)
+            logging.info(f"Input schema {input_schema} loaded successfully!")
+            for generator in GENERATORS:
+                output_dir = GENS_PATH / generator
+                output_dir.mkdir(parents=True, exist_ok=True)
+                logging.info(f"Proceeding with WoT resource generation")
+                run_generator(linkml_schema_view, generator, output_dir)
+        except yaml.YAMLError as e:
+            logging.info(f"LinkML schema validation failed: {e}")
+        if generate_docs:
+            logging.info(f"Generating documentation locally as markdown files...")
+            generate_documentation()
+        if serve_docs:
+            logging.info(f"Serving documentation...")
+            serve_documentation()
+
+
+if __name__ == "__main__":
+    main()

src/wotis/post_processors/jsonld_context_postprocessor.py

@@ -0,0 +1,93 @@
+import json
+import logging
+
+from linkml_runtime.linkml_model.meta import AnonymousSlotExpression
+from linkml_runtime.utils.schemaview import SchemaView
+
+XSD_NS = "http://www.w3.org/2001/XMLSchema#"
+
+
+def process_langstring_conditions(slot, context_entry):
+    """
+    Handles 'langString' and 'exactly_one_of' language conditions for a slot.
+    """
+    is_langstring = slot.range == 'langString'
+    is_exactly_one_language = (
+            slot.range is None and
+            any(
+                expr.range == 'langString'
+                for expr in slot.exactly_one_of if isinstance(expr, AnonymousSlotExpression)
+            )
+    )
+    if is_langstring or (is_exactly_one_language and isinstance(context_entry, dict)):
+        context_entry['@container'] = '@language'
+        context_entry.pop('@type', None)
+    return context_entry
+
+
+def process_multivalued_slots(slot, context_entry):
+    """
+    Handles multivalued slots and applies the '@set' container.
+    """
+    if slot.multivalued and isinstance(context_entry, dict) and '@container' not in context_entry:
+        context_entry['@container'] = '@set'
+        context_entry.pop('@type', None)
+    return context_entry
+
+
+def process_inlined_slot(slot, context_entry):
+    """
+    Processes slots that are both inlined, multi-valued and have a instantiates fields.
+    """
+    if slot.inlined and slot.multivalued and not slot.inlined_as_list and isinstance(context_entry, dict):
+        context_entry['@container'] = '@index'
+        context_entry['@type'] = '@id'
+        if slot.instantiates:
+            context_entry['@index'] = slot.instantiates
+    return context_entry
+
+
+def process_exactly_one_of(slot, context_entry):
+    """
+    Handles the 'exactly_one_of' condition for a slot, determining its type.
+    """
+    if hasattr(slot, 'exactly_one_of') and slot.exactly_one_of:
+        ranges = [opt['range'] for opt in slot.exactly_one_of if 'range' in opt]
+        if len(set(ranges)) == 1:
+            context_entry["@type"] = f"{XSD_NS}:{ranges[0]}"
+        else:
+            logging.warning(f"Warning: Slot {slot.name} has different ranges")
+    return context_entry
+
+
+def post_process_jsonld_context(schema_view: SchemaView, serialized_schema: str) -> str:
+    """
+    Post-processes a JSON-LD context generated from the default LinkML generators.
+    """
+    logging.info(f"Proceeding with JSON-LD Context postprocessor")
+    serialized_schema_json = json.loads(serialized_schema)
+    generated_context = serialized_schema_json.get('@context', {})
+    default_range = schema_view.schema.default_range if schema_view.schema.default_range else "string"
+    for slot in schema_view.all_slots().values():
+        slot_name = slot.name
+        context_entry = generated_context.get(slot_name, {})
+        # Process language conditions
+        context_entry = process_langstring_conditions(slot, context_entry)
+        # Add @language for LinkML in_language slot
+        if slot.in_language and isinstance(context_entry, dict):
+            context_entry['@language'] = slot.in_language
+        # Handle slots that are dict objects
+        context_entry = process_inlined_slot(slot, context_entry)
+        # Handle the @id for slots with exactly_one_of field
+        context_entry = process_exactly_one_of(slot, context_entry)
+        # Handles default range type
+        if slot.range == default_range:
+            context_entry["@type"] = f"{XSD_NS}:{default_range}"
+        generated_context[slot_name] = context_entry
+
+    # Intentionally handle multivalued slots separately after other processing
+    for slot in schema_view.all_slots().values():
+        context_entry = generated_context.get(slot.name, {})
+        context_entry = process_multivalued_slots(slot, context_entry)
+        generated_context[slot.name] = context_entry
+    return json.dumps(serialized_schema_json, indent=3)