[ENH] Add Learnwares Container

learnware/client/container.py

@@ -2,9 +2,12 @@ import os
 import pickle
 import tempfile
 import shortuuid
+from concurrent.futures import ProcessPoolExecutor
 
+from typing import List
 from .utils import system_execute, install_environment, remove_enviroment
 from ..config import C
+from ..learnware import Learnware
 from ..model.base import BaseModel
 
 from ..logger import get_module_logger
@@ -15,24 +18,23 @@ logger = get_module_logger(module_name="client_container")
 
 class ModelEnvContainer(BaseModel):
     def __init__(self, model_config: dict, learnware_zippath: str):
-        """The initialization method for base model
-        """
 
-        self.model_script = os.path.join(C.package_path, "learnware", "client", "run_model.py")
+        self.model_script = os.path.join(C.package_path, "client", "scripts", "run_model.py")
         self.model_config = model_config
-        self.conda_env = f"learnware_{shortuuid.uuid()}"
+        self.conda_env = f'learnware_{shortuuid.uuid()}'
         self.learnware_zippath = learnware_zippath
-        install_environment(learnware_zippath, self.conda_env)
 
+    def init_env_and_metadata(self):
+        install_environment(self.learnware_zippath, self.conda_env)
         with tempfile.TemporaryDirectory(prefix="learnware_") as tempdir:
             output_path = os.path.join(tempdir, "output.pkl")
             model_path = os.path.join(tempdir, "model.pkl")
 
             with open(model_path, "wb") as model_fp:
-                pickle.dump(model_config, model_fp)
+                pickle.dump(self.model_config, model_fp)
 
             system_execute(
-                f"conda run --no-capture-output python3 {self.model_script} --model-path {model_path} --output-path {output_path}"
+                ["conda", "run", "-n", f"{self.conda_env}", "--no-capture-output", "python3", f"{self.model_script}", f"--model-path", f"{model_path}", "--output-path", f"{output_path}"]
             )
 
             with open(output_path, "rb") as output_fp:
@@ -40,11 +42,12 @@ class ModelEnvContainer(BaseModel):
 
         if output_results["status"] != "success":
             raise output_results["error_info"]
-
         input_shape = output_results["metadata"]["input_shape"]
         output_shape = output_results["metadata"]["output_shape"]
-
         super(ModelEnvContainer, self).__init__(input_shape, output_shape)
+        
+    def remove_env(self):
+        remove_enviroment(self.conda_env)
 
     def run_model_with_script(self, method, **kargs):
         with tempfile.TemporaryDirectory(prefix="learnware_") as tempdir:
@@ -59,16 +62,16 @@ class ModelEnvContainer(BaseModel):
                 pickle.dump({"method": method, "kargs": kargs}, input_fp)
 
             system_execute(
-                f"conda run --no-capture-output python3 {self.model_script} --model-path {model_path} --input-path {input_path} --output-path {output_path}"
+                ["conda", "run", "-n", f"{self.conda_env}", "--no-capture-output", "python3", f"{self.model_script}", f"--model-path", f"{model_path}", f"--input-path", f"{input_path}", f"--output-path", "{output_path}"]
             )
 
             with open(output_path, "rb") as output_fp:
                 output_results = pickle.load(output_fp)
-
+                
         if output_results["status"] != "success":
             raise output_results["error_info"]
-
-        return output_results[output_results]
+        
+        return output_results[method]
 
     def fit(self, X, y):
         self.run_model_with_script("fit", X=X, y=y)
@@ -76,8 +79,52 @@ class ModelEnvContainer(BaseModel):
     def predict(self, X):
         return self.run_model_with_script("predict", X=X)
 
-    def finetune(self, X, y):
+    def finetune(self, X, y) -> None:
         self.run_model_with_script("finetune", X=X, y=y)
 
-    def __del__(self):
-        remove_enviroment(self.conda_env)
+
+class LearnwaresContainer:
+    
+    def __init__(self, learnware_list: List[Learnware], learnware_zippaths: List[str]):
+        """The initializaiton method for base reuser
+
+        Parameters
+        ----------
+        learnware_list : List[Learnware]
+            The learnware list to reuse and make predictions
+        """
+        assert all([isinstance(_learnware.get_model(), dict) for _learnware in learnware_list]), "the learnwre model should not be instantiated for reuser with containter"
+        self.learnware_list = [
+            Learnware(_learnware.id, ModelEnvContainer(_learnware.get_model(), _zippath), _learnware.get_specification()) for _learnware, _zippath in zip(learnware_list, learnware_zippaths)
+        ]
+    
+    @staticmethod
+    def _initialize_model_container(model: ModelEnvContainer):
+        try:
+            model.init_env_and_metadata()
+        except Exception as e:
+            logger.warning(f"fail to initialize model container, due to {e}")
+            pass
+    
+    @staticmethod
+    def _destroy_model_container(model: ModelEnvContainer):
+        try:
+            model.remove_env()
+        except Exception as e:
+            logger.warning(f"fail to destroy model container, due to {e}")
+            pass
+        
+    def __enter__(self):
+        model_list = [_learnware.get_model() for _learnware in self.learnware_list]
+        with ProcessPoolExecutor(max_workers=max(os.cpu_count() // 2, 1)) as executor:
+            executor.map(self._initialize_model_container, model_list)
+        return self
+    
+    def __exit__(self, type, value, trace):
+        model_list = [_learnware.get_model() for _learnware in self.learnware_list]
+        with ProcessPoolExecutor(max_workers=max(os.cpu_count() // 2, 1)) as executor:
+            executor.map(self._destroy_model_container, model_list)
+        return self
+    
+    def get_learnware_list_with_container(self):
+        return self.learnware_list

learnware/client/package_utils.py

@@ -1,6 +1,4 @@
-import os
 import yaml
-import time
 import subprocess
 from typing import List, Tuple
 
@@ -14,7 +12,7 @@ def try_to_run(args, timeout=5, retry=5):
     sucess = False
     for i in range(retry):
         try:
-            subprocess.check_call(args=args, timeout=timeout)
+            subprocess.check_call(args=args, timeout=timeout, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
             sucess = True
             break
         except subprocess.TimeoutExpired as e:
@@ -79,7 +77,6 @@ def filter_nonexist_pip_packages(packages: list) -> Tuple[List[str], List[str]]:
     for package in packages:
         try:
             # os.system("python3 -m pip index versions {0}".format(package))
-            logger.info("check package existence: {0}".format(package))
             try_to_run(args=["python3", "-m", "pip", "index", "versions", package], timeout=5)
             exist_packages.append(package)
         except Exception as e:

learnware/client/scripts/run_model.py

@@ -6,7 +6,7 @@ from learnware.utils import get_module_by_module_path
 
 def run_model(model_path, input_path, output_path):
     output_results = {"status": "success"}
-
+    
     try:
         with open(model_path, "rb") as model_file:
             model_config = pickle.load(file=model_file)
@@ -30,8 +30,9 @@ def run_model(model_path, input_path, output_path):
     except Exception as e:
         output_results["status"] = "fail"
         output_results["error_info"] = e
+        raise e
 
-    with open(output_path, "rb") as output_file:
+    with open(output_path, "wb") as output_file:
         pickle.dump(output_results, output_file)
 
 
@@ -47,4 +48,4 @@ if __name__ == "__main__":
     input_path = args.input_path
     output_path = args.output_path
 
-    print(model_path, input_path, output_path)
+    run_model(model_path, input_path, output_path)

learnware/client/utils.py

@@ -1,18 +1,22 @@
 import os
 import zipfile
 import tempfile
+import subprocess
 
 from ..logger import get_module_logger
 from .package_utils import filter_nonexist_conda_packages_file, filter_nonexist_pip_packages_file
 
 logger = get_module_logger(module_name="client_utils")
 
-
-def system_execute(command):
-    retcd: int = os.system(command=command)
-    if retcd != 0:
-        raise RuntimeError(f"Command {command} failed with return code {retcd}")
-
+def system_execute(args):
+    try:
+        com_process = subprocess.run(args, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, check=True)
+    except subprocess.CalledProcessError as err:
+        print(com_process.stderr)
+        raise err
+    
+def remove_enviroment(conda_env):
+    system_execute(args=["conda", "env", "remove", "-n", F"{conda_env}"])
 
 def install_environment(zip_path, conda_env):
     """Install environment of a learnware
@@ -31,29 +35,33 @@ def install_environment(zip_path, conda_env):
     """
     with tempfile.TemporaryDirectory(prefix="learnware_") as tempdir:
         with zipfile.ZipFile(file=zip_path, mode="r") as z_file:
-            logger.info(f"zip_file namelist: {z_file.namelist}")
+            logger.info(f"zip_file namelist: {z_file.namelist()}")
             if "environment.yaml" in z_file.namelist():
                 z_file.extract(member="environment.yaml", path=tempdir)
                 yaml_path: str = os.path.join(tempdir, "environment.yaml")
                 yaml_path_filter: str = os.path.join(tempdir, "environment_filter.yaml")
+                logger.info(f"checking the avaliabe conda packages for {conda_env}")
                 filter_nonexist_conda_packages_file(yaml_file=yaml_path, output_yaml_file=yaml_path_filter)
                 # create environment
-                system_execute(command=f"conda env update --name {conda_env} --file {yaml_path_filter}")
+                logger.info(f"create and update conda env [{conda_env}] according to .yaml file")
+                system_execute(args=["conda", "env", "update", "--name", f"{conda_env}", "--file", f"{yaml_path_filter}"])
 
             elif "requirements.txt" in z_file.namelist():
                 z_file.extract(member="requirements.txt", path=tempdir)
                 requirements_path: str = os.path.join(tempdir, "requirements.txt")
                 requirements_path_filter: str = os.path.join(tempdir, "requirements_filter.txt")
+                logger.info(f"checking the avaliabe pip packages for {yaml_path}")
                 filter_nonexist_pip_packages_file(
                     requirements_file=requirements_path, output_file=requirements_path_filter
                 )
-                system_execute(command=f"conda create --name {conda_env}")
+                logger.info(f"create empty conda env [{conda_env}]")
+                system_execute(args=["conda", "create", "--name", f"{conda_env}", "python=3.8"])
+                logger.info(f"install pip requirements for conda env [{conda_env}]")
                 system_execute(
-                    command=f"conda run --no-capture-output python3 -m pip install -r {requirements_path_filter}"
+                    args=["conda", "run", "--no-capture-output", "python3", "-m", "pip", "install", "-r", f"{requirements_path_filter}"]
                 )
             else:
                 raise Exception("Environment.yaml or requirements.txt not found in the learnware zip file.")
 
-
-def remove_enviroment(conda_env):
-    system_execute(command=f"conda env remove -n {conda_env}")
+    logger.info(f"install learnware package for conda env [{conda_env}]")
+    system_execute(args=["conda", "run", "--no-capture-output", "python3", "-m",  "pip",  "install", "learnware"])

learnware/config.py

@@ -64,11 +64,12 @@ LEARNWARE_ZIP_POOL_PATH = os.path.join(LEARNWARE_POOL_PATH, "zips")
 LEARNWARE_FOLDER_POOL_PATH = os.path.join(LEARNWARE_POOL_PATH, "learnwares")
 
 DATABASE_PATH = os.path.join(ROOT_DIRPATH, "database")
-
+STDOUT_PATH = os.path.join(ROOT_DIRPATH, "stdout")
 
 # TODO: Delete them later
 os.makedirs(ROOT_DIRPATH, exist_ok=True)
 os.makedirs(DATABASE_PATH, exist_ok=True)
+os.makedirs(STDOUT_PATH, exist_ok=True)
 
 semantic_config = {
     "Data": {"Values": ["Table", "Image", "Video", "Text", "Audio"], "Type": "Class",},  # Choose only one class
@@ -119,6 +120,7 @@ semantic_config = {
 _DEFAULT_CONFIG = {
     "root_path": ROOT_DIRPATH,
     "package_path": PACKAGE_DIRPATH,
+    "stdout_path": STDOUT_PATH,
     "logging_level": logging.INFO,
     "logging_outfile": None,
     "semantic_specs": semantic_config,

learnware/learnware/__init__.py

@@ -47,6 +47,7 @@ def get_learnware_from_dirpath(id: str, semantic_spec: dict, learnware_dirpath:
             yaml_config = read_yaml_to_dict(os.path.join(learnware_dirpath, C.learnware_folder_config["yaml_file"]))
         except FileNotFoundError:
             yaml_config = {}
+            raise
 
     if "name" in yaml_config:
         learnware_config["name"] = yaml_config["name"]

learnware/learnware/reuse.py

@@ -4,7 +4,7 @@ import numpy as np
 import geatpy as ea
 
 # import tensorflow as tf
-from typing import Tuple, Any, List, Union, Dict
+from typing import List
 from cvxopt import matrix, solvers
 from lightgbm import LGBMClassifier
 from scipy.special import softmax

learnware/market/easy.py

@@ -149,6 +149,7 @@ class EasyMarket(BaseMarket):
         except Exception as e:
             logger.exception
             logger.warning(f"The learnware [{learnware.id}] prediction is not avaliable! Due to {repr(e)}")
+            raise e
             return cls.NONUSABLE_LEARNWARE
 
         return cls.USABLE_LEARWARE

tests/test_learnware_client/test_reuse.py

@@ -1,6 +1,10 @@
+import zipfile
+import numpy as np
+
 from learnware.learnware import get_learnware_from_dirpath, Learnware
 from learnware.market import EasyMarket
 from learnware.client.container import ModelEnvContainer
+from learnware.learnware.reuse import AveragingReuser
 
 if __name__ == "__main__":
     semantic_specification = dict()
@@ -11,11 +15,36 @@ if __name__ == "__main__":
     semantic_specification["Name"] = {"Type": "String", "Values": "test"}
     semantic_specification["Description"] = {"Type": "String", "Values": "test"}
     
-    zip_path = '/home/bixd/workspace/learnware/Learnware/tests/test_workflow/learnware_pool/svm_0.zip'
+    zip_paths = [
+        '/home/bixd/workspace/learnware/Learnware/tests/test_learnware_client/rf_tic.zip',
+        '/home/bixd/workspace/learnware/Learnware/tests/test_learnware_client/svc_tic.zip',
+    ]
+    dir_paths = [
+        '/home/bixd/workspace/learnware/Learnware/tests/test_learnware_client/rf_tic',
+        '/home/bixd/workspace/learnware/Learnware/tests/test_learnware_client/svc_tic',
+    ]
+    
+    learnware_list = []
+    for id, (zip_path, dir_path) in enumerate(zip(zip_paths, dir_paths)):
+        with zipfile.ZipFile(zip_path, "r") as z_file:
+            z_file.extractall(dir_path)
+            
+        learnware = get_learnware_from_dirpath(f'test_id{id}', semantic_specification, dir_path)
+
+        model = ModelEnvContainer(learnware.get_model(), zip_path)
+        model.init_env_and_metadata()
+        
+        env_leanware = Learnware(id=learnware.id, model=model, specification=learnware.get_specification())
+            
+        learnware_list.append(env_leanware)
+        
+        print('check:', EasyMarket.check_learnware(env_leanware))
     
-    learnware = get_learnware_from_dirpath('test_id', semantic_specification, zip_path)
+    reuser = AveragingReuser(learnware_list, mode='vote')
+    input_array = np.random.randint(0, 3, size=(20, 9))
+    print(reuser.predict(input_array).argmax(axis=1))
     
-    env_leanware = Learnware(id=learnware.id, model=ModelEnvContainer(learnware.get_model(), zip_path), specification=learnware.get_specification())
+    for id, ind_learner in enumerate(learnware_list):
+        print(f"learner_{id}", reuser.predict(input_array).argmax(axis=1))
     
-    print('check', EasyMarket.check_learnware(env_leanware))
     

tests/test_learnware_client/test_reuse2.py

@@ -0,0 +1,46 @@
+import zipfile
+import numpy as np
+
+from learnware.learnware import get_learnware_from_dirpath, Learnware
+from learnware.market import EasyMarket
+from learnware.client.container import ModelEnvContainer, LearnwaresContainer
+from learnware.learnware.reuse import AveragingReuser
+
+if __name__ == "__main__":
+    semantic_specification = dict()
+    semantic_specification["Data"] = {"Type": "Class", "Values": ["Text"]}
+    semantic_specification["Task"] = {"Type": "Class", "Values": ["Ranking"]}
+    semantic_specification["Library"] = {"Type": "Class", "Values": ["Scikit-learn"]}
+    semantic_specification["Scenario"] = {"Type": "Tag", "Values": "Financial"}
+    semantic_specification["Name"] = {"Type": "String", "Values": "test"}
+    semantic_specification["Description"] = {"Type": "String", "Values": "test"}
+    
+    zip_paths = [
+        '/home/bixd/workspace/learnware/Learnware/tests/test_learnware_client/rf_tic.zip',
+        '/home/bixd/workspace/learnware/Learnware/tests/test_learnware_client/svc_tic.zip',
+    ]
+    dir_paths = [
+        '/home/bixd/workspace/learnware/Learnware/tests/test_learnware_client/rf_tic',
+        '/home/bixd/workspace/learnware/Learnware/tests/test_learnware_client/svc_tic',
+    ]
+    
+    
+    learnware_list = []
+    for id, (zip_path, dir_path) in enumerate(zip(zip_paths, dir_paths)):
+        with zipfile.ZipFile(zip_path, "r") as z_file:
+            z_file.extractall(dir_path)
+            
+        learnware = get_learnware_from_dirpath(f'test_id{id}', semantic_specification, dir_path)
+        learnware_list.append(learnware)
+    
+    with LearnwaresContainer(learnware_list, zip_paths) as env_container:
+        learnware_list = env_container.get_learnware_list_with_container()
+    
+        reuser = AveragingReuser(learnware_list, mode='vote')
+        input_array = np.random.randint(0, 3, size=(20, 9))
+        print(reuser.predict(input_array).argmax(axis=1))
+    
+        for id, ind_learner in enumerate(learnware_list):
+            print(f"learner_{id}", reuser.predict(input_array).argmax(axis=1))
+    
+