Add metadata using robofair

pyproject.toml

@@ -16,7 +16,8 @@ dependencies = [
     "pyyaml",
     "floorplan_dsl@git+https://github.com/secorolab/FloorPlan-DSL",
     "transforms3d",
-    "ifcld@git+https://github.com/secorolab/ifcld"
+    "ifcld@git+https://github.com/secorolab/ifcld",
+    "robofair@git+https://github.com/secorolab/robofair"
 ]
 requires-python = ">= 3.11"
 readme = "README.md"

src/fpm/cli.py

@@ -7,6 +7,9 @@
     fpm_prov_generation_graph,
     artefact_prov_generation_graph,
     var_prov_generation_graph,
+    var_prov_metadata,
+    artefact_prov_metadata,
+    jsonld_prov_metadata,
 )
 from fpm.graph import build_graph_from_directory, get_floorplan_model_name
 from fpm.generators.gazebo import gazebo_world, door_object_models
@@ -157,6 +160,7 @@ def transform(ctx, model_path, output_path, **kwargs):
     except Exception as e:
         logger.error(f"Error transforming model: {e}")
 
+    jsonld_prov_metadata(model_path, res)
     if kwargs.get("prov"):
         fpm_prov_generation_graph(model, model_path, res, output_path, **kwargs)
 
@@ -248,6 +252,8 @@ def variation(ctx, model_path, variations, seed, output_path, **kwargs):
     except Exception as e:
         logger.error(f"Error generating variations: {e}")
 
+    var_prov_metadata(model_path, f, res)
+
     if kwargs.get("prov"):
         var_prov_generation_graph(model_path, f, res, output_path, **kwargs)
 
@@ -366,12 +372,13 @@ def mesh(ctx, **kwargs):
     """Generate a 3D-mesh in STL or gltF 2.0 format"""
     output_file = get_3d_mesh(**ctx.obj, **ctx.parent.params, **kwargs)
 
+    artefact_prov_metadata(ctx.parent.params.get("inputs"), output_file)
     prov = ctx.parent.params.get("prov")
     if prov:
         artefact_prov_generation_graph(
             ctx.obj.get("model_name"),
             ctx.parent.params.get("inputs"),
-            [output_file],
+            output_file,
             "3D-Mesh",
             ctx.parent.params.get("base_path"),
             model_base_iri=ctx.parent.params.get("model_base_iri"),
@@ -452,6 +459,9 @@ def gazebo(ctx, **kwargs):
     files = gazebo_world(**ctx.obj, **ctx.parent.params, **kwargs)
     output_files.extend(files)
 
+    artefact_prov_metadata(
+        ctx.parent.params.get("inputs"), output_files, ignored_extensions=[".jpg"]
+    )
     prov = ctx.parent.params.get("prov")
     if prov:
         artefact_prov_generation_graph(
@@ -561,6 +571,7 @@ def occ_grid(ctx, **kwargs):
     logger.debug("Arguments: %s", kwargs)
     output_files = get_occ_grid(**ctx.obj, **ctx.parent.params, **kwargs)
 
+    artefact_prov_metadata(ctx.parent.params.get("inputs"), output_files)
     prov = ctx.parent.params.get("prov")
     if prov:
         artefact_prov_generation_graph(

src/fpm/generators/prov.py

@@ -1,9 +1,13 @@
 import os
 import json
-
+import glob
 import datetime
+import logging
 
 from fpm.utils import load_template, save_file, get_output_path
+from robofair.metadata import FileMetadata
+
+logger = logging.getLogger("floorplan.generators.prov")
 
 
 def fpm_prov_generation_graph(model, model_path, generated_files, base_path, **kwargs):
@@ -28,6 +32,12 @@ def fpm_prov_generation_graph(model, model_path, generated_files, base_path, **k
     return save_file(output_path, file_name, contents)
 
 
+def jsonld_prov_metadata(fpm_file, generated_files, **kwargs):
+    fpm_file_metadata(fpm_file)
+    for f in generated_files:
+        gen_file_metadata(f, [fpm_file])
+
+
 def artefact_prov_generation_graph(
     env_id, source_files, generated_files, artefact_type, base_path, **kwargs
 ):
@@ -52,6 +62,48 @@ def artefact_prov_generation_graph(
     return save_file(output_path, file_name, contents)
 
 
+def artefact_prov_metadata(source_files, generated_files, **kwargs):
+    _source_files = []
+    for d in source_files:
+        _source_files.extend(glob.glob(os.path.join(d, "*.json")))
+
+    for f in generated_files:
+        if os.path.basename(os.path.basename(f)) in kwargs.get(
+            "ignored_extensions", []
+        ):
+            logger.debug(f"Skipping {f} metadata")
+            continue
+
+        gen_file_metadata(f, _source_files)
+
+
+def gen_file_metadata(f, source_files):
+    f_mdata = FileMetadata(f)
+    mdata = dict(
+        created_at=f_mdata.missing_metadata.get("updated_at"),
+        attributed_to="https://purl.org/secorolab/scenery_builder",
+        derived_from=[os.path.relpath(sf, os.path.dirname(f)) for sf in source_files],
+    )
+    f_mdata.add_missing_metadata(mdata, save=True)
+
+
+def fpm_file_metadata(fpm_file):
+    fpm_mdata = FileMetadata(fpm_file)
+    if fpm_mdata.metadata_file is None:
+        fpm_mdata.save_missing_metadata()
+
+
+def var_prov_metadata(var_file, fpm_file, generated_files, **kwargs):
+    for f in generated_files:
+        gen_file_metadata(f, [var_file])
+
+    fpm_file_metadata(fpm_file)
+
+    var_mdata = FileMetadata(var_file)
+    var_ref_fpm = {"references": os.path.relpath(fpm_file, os.path.dirname(var_file))}
+    var_mdata.add_missing_metadata(var_ref_fpm, save=True)
+
+
 def var_prov_generation_graph(var_file, fpm_file, generated_files, base_path, **kwargs):
     gen_time = datetime.datetime.now().isoformat()
     template = load_template("prov/var-file-generation.json.jinja")