From 11729f4df2bcd36342f3933fb05d23a0ae908e94 Mon Sep 17 00:00:00 2001 From: bxdd Date: Wed, 11 Oct 2023 14:22:07 +0800 Subject: [PATCH] [ENH] Add Learnwares Container --- learnware/client/container.py | 79 +++++++++++++++++----- learnware/client/package_utils.py | 5 +- learnware/client/scripts/run_model.py | 7 +- learnware/client/utils.py | 34 ++++++---- learnware/config.py | 4 +- learnware/learnware/__init__.py | 1 + learnware/learnware/reuse.py | 2 +- learnware/market/easy.py | 1 + tests/test_learnware_client/test_reuse.py | 37 ++++++++-- tests/test_learnware_client/test_reuse2.py | 46 +++++++++++++ 10 files changed, 174 insertions(+), 42 deletions(-) create mode 100644 tests/test_learnware_client/test_reuse2.py diff --git a/learnware/client/container.py b/learnware/client/container.py index c0df38e..c59b4f2 100644 --- a/learnware/client/container.py +++ b/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) \ No newline at end of file + +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 \ No newline at end of file diff --git a/learnware/client/package_utils.py b/learnware/client/package_utils.py index 3caff09..0088bb4 100644 --- a/learnware/client/package_utils.py +++ b/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: diff --git a/learnware/client/scripts/run_model.py b/learnware/client/scripts/run_model.py index a284184..0f735f3 100644 --- a/learnware/client/scripts/run_model.py +++ b/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) \ No newline at end of file diff --git a/learnware/client/utils.py b/learnware/client/utils.py index e619d95..5e315f2 100644 --- a/learnware/client/utils.py +++ b/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"]) diff --git a/learnware/config.py b/learnware/config.py index d6cb110..137f15b 100644 --- a/learnware/config.py +++ b/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, diff --git a/learnware/learnware/__init__.py b/learnware/learnware/__init__.py index dfde4d5..70e35d6 100644 --- a/learnware/learnware/__init__.py +++ b/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"] diff --git a/learnware/learnware/reuse.py b/learnware/learnware/reuse.py index ae75cb8..66bc148 100644 --- a/learnware/learnware/reuse.py +++ b/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 diff --git a/learnware/market/easy.py b/learnware/market/easy.py index 15f7434..550df68 100644 --- a/learnware/market/easy.py +++ b/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 diff --git a/tests/test_learnware_client/test_reuse.py b/tests/test_learnware_client/test_reuse.py index e74ddbe..680622d 100644 --- a/tests/test_learnware_client/test_reuse.py +++ b/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)) \ No newline at end of file diff --git a/tests/test_learnware_client/test_reuse2.py b/tests/test_learnware_client/test_reuse2.py new file mode 100644 index 0000000..0dcf293 --- /dev/null +++ b/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)) + + \ No newline at end of file