Merge branch 'bixd/dev' of https://github.com/Learnware-LAMDA/Learnware into bixd/dev

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,35 @@ 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.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,12 +54,13 @@ 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:
             input_path = os.path.join(tempdir, "input.pkl")
@@ -59,7 +74,21 @@ 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",
+                    f"{output_path}",
+                ]
             )
 
             with open(output_path, "rb") as output_fp:
@@ -68,7 +97,7 @@ class ModelEnvContainer(BaseModel):
         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 +105,55 @@ 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)
+
+    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

@@ -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,6 +1,7 @@
 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
@@ -8,10 +9,18 @@ from .package_utils import filter_nonexist_conda_packages_file, filter_nonexist_
 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, timeout=None):
+    try:
+        com_process = subprocess.run(
+            args, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, check=True, timeout=timeout
+        )
+    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):
@@ -31,29 +40,45 @@ 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

@@ -3,7 +3,7 @@ import random
 import numpy as np
 import geatpy as ea
 
-from typing import Tuple, Any, List, Union, Dict
+from typing import List
 from cvxopt import matrix, solvers
 from lightgbm import LGBMClassifier, early_stopping
 from scipy.special import softmax
@@ -665,7 +665,7 @@ class EnsemblePruningReuser(BaseReuser):
                 pred_y = pred_y.detach().cpu().numpy()
             if not isinstance(pred_y, np.ndarray):
                 raise TypeError(f"Model output must be np.ndarray or torch.Tensor")
-            
+
             if len(pred_y.shape) == 1:
                 pred_y = pred_y.reshape(-1, 1)
             elif len(pred_y.shape) == 2:
@@ -674,9 +674,9 @@ class EnsemblePruningReuser(BaseReuser):
             else:
                 raise ValueError("Model output must be a 1D or 2D vector")
             preds.append(pred_y)
-        
+
         return np.concatenate(preds, axis=1)
-    
+
     def fit(self, val_X: np.ndarray, val_y: np.ndarray, maxgen: int = 500):
         """Ensemble pruning based on the validation set
 

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

learnware/utils.py

@@ -50,6 +50,6 @@ def save_dict_to_yaml(dict_value: dict, save_path: str):
 
 def read_yaml_to_dict(yaml_path: str):
     """load yaml file into dict object"""
-    with open(yaml_path, 'r') as file:
+    with open(yaml_path, "r") as file:
         dict_value = yaml.load(file.read(), Loader=yaml.FullLoader)
         return dict_value

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.client.container import ModelEnvContainer, LearnwaresContainer
+from learnware.learnware.reuse import AveragingReuser
 
 if __name__ == "__main__":
     semantic_specification = dict()
@@ -10,12 +14,29 @@ if __name__ == "__main__":
     semantic_specification["Scenario"] = {"Type": "Tag", "Values": "Financial"}
     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'
-    
-    learnware = get_learnware_from_dirpath('test_id', semantic_specification, zip_path)
-    
-    env_leanware = Learnware(id=learnware.id, model=ModelEnvContainer(learnware.get_model(), zip_path), specification=learnware.get_specification())
-    
-    print('check', EasyMarket.check_learnware(env_leanware))
-    
+
+    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))

tests/test_workflow/test_workflow.py

@@ -187,7 +187,7 @@ class TestAllWorkflow(unittest.TestCase):
         # Use averaging ensemble reuser to reuse the searched learnwares to make prediction
         reuse_ensemble = AveragingReuser(learnware_list=mixture_learnware_list, mode="vote_by_prob")
         ensemble_predict_y = reuse_ensemble.predict(user_data=data_X)
-        
+
         # Use ensemble pruning reuser to reuse the searched learnwares to make prediction
         reuse_ensemble = EnsemblePruningReuser(learnware_list=mixture_learnware_list, mode="classification")
         reuse_ensemble.fit(train_X[-200:], train_y[-200:])