[MNT] add api to update semantic specification of learnware

learnware/learnware/base.py

@@ -68,7 +68,10 @@ class Learnware:
 
     def update_stat_spec(self, name, new_stat_spec: BaseStatSpecification):
         self.specification.update_stat_spec(name, new_stat_spec)
-
+    
+    def update_semantic_spec(self, new_semantic_spec: dict):
+        self.specification.update_semantic_spec(new_semantic_spec)
+        
     def update(self):
         # Empty Interface.
         raise NotImplementedError("'update' Method is NOT Implemented.")

learnware/market/database_ops.py

@@ -117,6 +117,17 @@ class DatabaseOperations(object):
             pass
         pass
 
+    def update_learnware_semantic_spec(self, learnware_id: str, semantic_spec: dict):
+        with self.engine.connect() as conn:
+            semantic_spec_str = json.dumps(semantic_spec)
+            conn.execute(
+                text("UPDATE tb_learnware SET semantic_spec=:semantic_spec WHERE id=:id;"),
+                dict(id=learnware_id, semantic_spec=semantic_spec_str),
+            )
+            conn.commit()
+            pass
+        pass
+
     def delete_learnware(self, id: str):
         with self.engine.connect() as conn:
             conn.execute(text("DELETE FROM tb_learnware WHERE id=:id;"), dict(id=id))

learnware/market/easy.py

@@ -8,6 +8,7 @@ import pandas as pd
 from cvxopt import solvers, matrix
 from typing import Tuple, Any, List, Union, Dict
 import traceback
+import json
 
 from .base import BaseMarket, BaseUserInfo
 from .database_ops import DatabaseOperations
@@ -16,6 +17,8 @@ from ..learnware import Learnware, get_learnware_from_dirpath
 from ..specification import RKMEStatSpecification, Specification
 from ..logger import get_module_logger
 from ..config import C as conf
+from .. import utils
+
 
 logger = get_module_logger("market", "INFO")
 
@@ -834,6 +837,26 @@ class EasyMarket(BaseMarket):
                 logger.warning("Learnware ID '%s' NOT Found!" % (ids))
                 return None
 
+    def update_learnware_semantic_spec(self, learnware_id: str, semantic_spec: dict) -> bool:
+        """Update Learnware semantic_spec
+        """
+
+        # update database
+        self.dbops.update_learnware_semantic_spec(learnware_id=learnware_id, semantic_spec=semantic_spec)
+        # update file
+        
+        folder_path = self.learnware_folder_list[learnware_id]
+        with open(os.path.join(folder_path, "semantic_specification.json"), "w") as f:
+            json.dump(semantic_spec, f)
+            pass
+        # update zip
+        zip_path = self.learnware_zip_list[learnware_id]
+        utils.zip_learnware_folder(folder_path, zip_path)
+
+        # update learnware
+        self.learnware_list[learnware_id].update_semantic_spec(semantic_spec)
+        pass
+
     def __len__(self):
         return len(self.learnware_list.keys())
 

learnware/utils.py

@@ -7,6 +7,7 @@ import importlib
 import importlib.util
 from typing import Union
 from types import ModuleType
+import zipfile
 from .logger import get_module_logger
 
 logger = get_module_logger("utils")
@@ -53,3 +54,20 @@ def read_yaml_to_dict(yaml_path: str):
     with open(yaml_path, "r") as file:
         dict_value = yaml.load(file.read(), Loader=yaml.FullLoader)
         return dict_value
+
+def zip_learnware_folder(path: str, output_name: str):
+    with zipfile.ZipFile(output_name, "w") as zip_ref:
+        for root, dirs, files in os.walk(path):
+            for file in files:
+                full_path = os.path.join(root, file)
+                if file.endswith(".pyc") or os.path.islink(full_path):
+                    continue
+                zip_ref.write(
+                    full_path, 
+                    arcname=os.path.relpath(
+                        full_path, 
+                        path))
+                pass
+            pass
+        pass
+    pass