[ENH] add test template

learnware/learnware/__init__.py

@@ -1,8 +1,8 @@
 import os
 import copy
+from typing import Optional
 
 from .base import Learnware
-
 from .utils import get_stat_spec_from_config
 from ..specification import Specification
 from ..utils import read_yaml_to_dict
@@ -12,7 +12,7 @@ from ..config import C
 logger = get_module_logger("learnware.learnware")
 
 
-def get_learnware_from_dirpath(id: str, semantic_spec: dict, learnware_dirpath, ignore_error=True) -> Learnware:
+def get_learnware_from_dirpath(id: str, semantic_spec: dict, learnware_dirpath, ignore_error=True) -> Optional[Learnware]:
     """Get the learnware object from dirpath, and provide the manage interface tor Learnware class
 
     Parameters

learnware/tests/module.py

@@ -8,3 +8,8 @@ def get_semantic_specification():
     semantic_specification["Name"] = {"Type": "String", "Values": "test"}
     semantic_specification["Description"] = {"Type": "String", "Values": "test"}
     return semantic_specification
+
+
+
+
+def get_requirements_file()

learnware/tests/templates/__init__.py

@@ -0,0 +1,76 @@
+import os
+import tempfile
+from shutil import copyfile
+from typing import List, Tuple, Union, Optional
+
+from ...utils import save_dict_to_yaml
+from ...config import C
+
+class LearnwareTemplate:
+    def __init__(self):
+        self.model_templates = {
+            "pickle": {
+                "class_name": 'PickleLoadedModel',
+                "template_path": os.path.join(C.package_path, "tests", "templates", "pickle_model.py")
+            }
+        }
+    
+    def generate_requirements(self, filepath, requirements: Optional[List[Union[Tuple[str, str, str], str]]] = None):
+        requirements = [] if requirements is None else requirements
+        operators = {"==", "~=", ">=", "<=", ">", "<"}
+        requirements_str = ""
+        for requirement in requirements:
+            if isinstance(requirement, str):
+                line_str = requirement.strip() + "\n"
+            elif isinstance(requirement, tuple):
+                assert requirement[1] in operators, f"The operator of requirements is not supported."
+                line_str = requirement[0].strip() + requirement[1].strip() + requirement[2].strip() + "\n"
+            else:
+                raise TypeError(f"requirement must be type str/tuple, rather than {type(requirement)}")
+            
+            requirements_str += line_str
+            
+        with open(filepath, "w") as fdout:
+            fdout.write(requirements_str)
+        
+    def generate_learnware_yaml(self, filepath, model_config: Optional[dict] = None, stat_spec_config: Optional[List[dict]] = None):
+        learnware_config = {}
+        if model_config is not None:
+            learnware_config["model"] = model_config
+        if stat_spec_config is not None:
+            learnware_config["stat_specifications"] = stat_spec_config
+
+        save_dict_to_yaml(learnware_config, filepath)
+        
+    
+    
+    def generate_learnware_zipfile(
+        self,
+        learnware_zippath: str,
+        model_template: str = "pickle",
+        model_kwargs: Optional[dict] = None,
+        stat_spec_config: Optional[List[dict]] = None,
+        requirements: Optional[List[Union[Tuple[str, str, str], str]]] = None,
+        **kwargs,
+    ):
+        with tempfile.TemporaryDirectory(suffix="learnware_template") as tempdir:
+            requirement_filepath = os.path.join(tempdir, "requirements.txt")
+            self.generate_requirements(requirement_filepath, requirements)
+            
+            model_filepath =  os.path.join(tempdir, "__init__.py")
+            copyfile(self.model_templates[model_template]["template_path"], model_filepath)
+            
+            learnware_yaml_filepath = os.path.join(tempdir, "requirements.txt")
+            model_config = {
+                "class_name": self.model_templates[model_template]["class_name"],
+                "kwargs": {} if model_kwargs is None else model_kwargs
+            }
+            self.generate_learnware_yaml(learnware_yaml_filepath, model_config, stat_spec_config)
+
+            if model_template == "pickle":
+                pickle_filepath = os.path.join(tempdir, model_config["kwargs"]["pickle_filepath"])
+                copyfile(kwargs["pickle_filepath"], pickle_filepath)
+                
+
+    def generate_template_semantic_spec(self):
+        pass

learnware/tests/templates/pickle_model.py

@@ -0,0 +1,31 @@
+import os
+import pickle
+import numpy as np
+from learnware.model.base import BaseModel
+
+class PickleLoadedModel(BaseModel):
+    
+    def __init__(
+        self,
+        input_shape,
+        output_shape,
+        pickle_filepath,
+        predict_method="predict",
+        fit_method="fit",
+        finetune_method="finetune",
+    ):
+        super(PickleLoadedModel, self).__init__(input_shape=input_shape, output_shape=output_shape)
+        with open(pickle_filepath, "rb") as fd:
+            self.model = pickle.load(fd)
+        self.predict_method = predict_method
+        self.fit_method = fit_method
+        self.finetune_method = finetune_method
+            
+    def predict(self, X: np.ndarray) -> np.ndarray:
+        return getattr(self.model, self.predict_method)(X)
+    
+    def fit(self, X: np.ndarray, y: np.ndarray):
+        getattr(self.model, self.fit_method)(X, y)
+
+    def finetune(self, X: np.ndarray, y: np.ndarray):
+        getattr(self.model, self.finetune_method)(X, y)

tests/test_reuse/test_averaging.py

@@ -0,0 +1,43 @@
+import os
+import json
+import string
+import random
+import torch
+import unittest
+import tempfile
+import numpy as np
+
+from learnware.specification import RKMETableSpecification, HeteroMapTableSpecification
+from learnware.specification import generate_stat_spec
+from learnware.market.heterogeneous.organizer import HeteroMap
+
+class TestAveragingReuse(unittest.TestCase):
+    
+    def setUp(self):
+        self.hetero_map = HeteroMap()
+        
+    def _test_hetero_spec(self, X):
+        rkme: RKMETableSpecification = generate_stat_spec(type="table", X=X)
+        hetero_spec = self.hetero_map.hetero_mapping(rkme_spec=rkme, features=dict())
+        with tempfile.TemporaryDirectory(prefix="learnware_") as tempdir:
+            rkme_path = os.path.join(tempdir, "rkme.json")
+            hetero_spec.save(rkme_path)
+
+            with open(rkme_path, "r") as f:
+                data = json.load(f)
+                assert data["type"] == "HeteroMapTableSpecification"
+
+            rkme2 = HeteroMapTableSpecification()
+            rkme2.load(rkme_path)
+            assert rkme2.type == "HeteroMapTableSpecification"
+        
+            
+    def test_hetero_rkme(self):
+        self._test_hetero_spec(np.random.uniform(-10000, 10000, size=(5000, 200)))
+        self._test_hetero_spec(np.random.uniform(-10000, 10000, size=(10000, 100)))
+        self._test_hetero_spec(np.random.uniform(-10000, 10000, size=(5, 20)))
+        self._test_hetero_spec(np.random.uniform(-10000, 10000, size=(1, 50)))
+        self._test_hetero_spec(np.random.uniform(-10000, 10000, size=(100, 150)))
+        
+if __name__ == "__main__":
+    unittest.main()

tests/test_workflow/learnware_example/example_init.py

@@ -8,7 +8,7 @@ class SVM(BaseModel):
     def __init__(self):
         super(SVM, self).__init__(input_shape=(64,), output_shape=(10,))
         dir_path = os.path.dirname(os.path.abspath(__file__))
-        self.model = joblib.load(os.path.join(dir_path, "svm.pkl"))
+        self.model = pickle.load(os.path.join(dir_path, "svm.pkl"))
 
     def fit(self, X: np.ndarray, y: np.ndarray):
         pass