From 50551d42656ebbc2deb2d1f80de1bdc06695cd71 Mon Sep 17 00:00:00 2001 From: bxdd Date: Thu, 26 Oct 2023 22:32:39 +0800 Subject: [PATCH] [MNT] maintain use_flags in easy market --- learnware/market/base.py | 192 ++++++------------ learnware/market/database_ops.py | 18 +- learnware/market/easy.py | 4 +- learnware/market/easy/checker.py | 2 + learnware/market/easy/database_ops.py | 31 +-- learnware/market/easy/organizer.py | 275 +++++++++++++++----------- 6 files changed, 241 insertions(+), 281 deletions(-) diff --git a/learnware/market/base.py b/learnware/market/base.py index 12c8f9f..2bec206 100644 --- a/learnware/market/base.py +++ b/learnware/market/base.py @@ -43,6 +43,10 @@ class BaseUserInfo: return self.stat_info.get(name, None) +class BaseSearchResult: + + pass + class LearnwareMarket: """Base interface for market, it provide the interface of search/add/detele/update learnwares""" @@ -62,145 +66,43 @@ class LearnwareMarket: self.learnware_searcher.reset(organizer=self.learnware_organizer) def reload_market(self, **kwargs) -> bool: - """Reload the market when server restared. - Returns - ------- - bool - A flag indicating whether the market is reload successfully. - """ self.learnware_organizer.reload_market(**kwargs) def check_learnware(self, learnware: Learnware, **kwargs) -> bool: - """Check the utility of a learnware - - Parameters - ---------- - learnware : Learnware - - Returns - ------- - bool - A flag indicating whether the learnware can be accepted. - """ return self.learnware_checker(learnware, **kwargs) def add_learnware(self, zip_path: str, semantic_spec: dict, **kwargs) -> Tuple[str, bool]: - """Add a learnware into the market. - - .. note:: - - Given a prediction of a certain time, all signals before this time will be prepared well. - - Parameters - ---------- - zip_path : str - Filepath for learnware model, a zipped file. - semantic_spec : dict - semantic_spec for new learnware, in dictionary format. - - Returns - ------- - Tuple[str, bool] - str indicating model_id, bool indicating whether the learnware is added successfully. - - Raises - ------ - FileNotFoundError - file for model or statistical specification not found - - """ return self.learnware_organizer.add_learnware(zip_path, semantic_spec, **kwargs) def search_learnware(self, user_info: BaseUserInfo, **kwargs) -> Tuple[Any, List[Learnware]]: - """Search Learnware based on user_info - - Parameters - ---------- - user_info : BaseUserInfo - user_info with emantic specifications and statistical information - - Returns - ------- - Tuple[Any, List[Any]] - return two items: - - - first is recommended combination, None when no recommended combination is calculated or statistical specification is not provided. - - second is a list of matched learnwares - """ - return self.learnware_searcher(user_info, **kwargs) - def delete_learnware(self, id: str, *args, **kwargs) -> bool: - """Delete a learnware from market - - Parameters - ---------- - id : str - id of learnware to be deleted - - Returns - ------- - bool - True if the target learnware is deleted successfully. - - Raises - ------ - Exception - Raise an excpetion when given id is NOT found in learnware list - """ + def delete_learnware(self, id: str, **kwargs) -> bool: return self.learnware_organizer.delete_learnware(id, **kwargs) def update_learnware(self, id: str, zip_path: str, semantic_spec: dict, **kwargs) -> bool: - """ - Update Learnware with id and content to be updated. - Empty interface. TODO - - Parameters - ---------- - id : str - id of target learnware. - """ return self.learnware_organizer.update_learnware(id, zip_path=zip_path, semantic_spec=semantic_spec, **kwargs) - def get_semantic_spec_list(self) -> dict: - """Return all semantic specifications available - - Returns - ------- - dict - All emantic specifications in dictionary format - - """ - raise NotImplementedError("get semantic spec list is not implemented") + def get_learnware_ids(self, top:int = None, **kwargs): + return self.learnware_organizer.get_learnware_ids(top, **kwargs) + - def get_learnware_by_ids(self, id: Union[str, List[str]]) -> Union[Learnware, List[Learnware]]: - """ - Get Learnware from market by id - - Parameters - ---------- - id : Union[str, List[str]] - Given one id or a list of ids as target. - - Returns - ------- - Union[Learnware, List[Learnware]] - Return a Learnware object or a list of Learnware objects based on the type of input param. - - - The returned items are search results. - - 'None' indicating the target id not found. - """ - return self.learnware_organizer.get_learnware_by_ids(id) + def get_learnwares(self, top:int = None, **kwargs): + return self.learnware_organizer.get_learnwares(top, **kwargs) + + def get_learnware_path_by_ids(self, ids: Union[str, List[str]], **kwargs) -> Union[Learnware, List[Learnware]]: + raise self.learnware_organizer.get_learnware_path_by_ids(ids, **kwargs) - def + def get_learnware_by_ids(self, id: Union[str, List[str]], **kwargs) -> Union[Learnware, List[Learnware]]: + return self.learnware_organizer.get_learnware_by_ids(id, **kwargs) class LearnwareOrganizer: - def __init__(self, market_id, organizer: 'LearnwareOrganizer' = None): - self.reset(market_id=market_id, organizer=organizer) + def __init__(self, market_id, checker: 'LearnwareChecker' = None): + self.reset(market_id=market_id, checker=checker) - def reset(self, market_id, organizer: 'LearnwareOrganizer' ): + def reset(self, market_id, checker: 'LearnwareChecker', **kwargs): self.market_id = market_id - self.organizer = organizer + self.organizer = checker def reload_market(self) -> bool: """Reload the learnware organizer when server restared. @@ -267,7 +169,6 @@ class LearnwareOrganizer: def update_learnware(self, id: str, zip_path: str, semantic_spec: dict, **kwargs) -> bool: """ Update Learnware with id and content to be updated. - Empty interface. TODO Parameters ---------- @@ -276,17 +177,6 @@ class LearnwareOrganizer: """ raise NotImplementedError("update learnware is Not Implemented") - def get_semantic_spec_list(self) -> dict: - """Return all semantic specifications available - - Returns - ------- - dict - All emantic specifications in dictionary format - - """ - raise NotImplementedError("get semantic spec list is not implemented") - def get_learnware_by_ids(self, id: Union[str, List[str]]) -> Union[Learnware, List[Learnware]]: """ Get Learnware from market by id @@ -324,11 +214,36 @@ class LearnwareOrganizer: """ raise NotImplementedError("get_learnware_path_by_ids is not implemented") - def get_learnware_ids(self, top:int = None): - if top is None: - return list(self.learnware_list.keys()) - else: - return list(self.learnware_list.keys())[:top] + def get_learnware_ids(self, top:int = None) -> List[str]: + """get the list of learnware ids + + Parameters + ---------- + top : int, optional + the first top element to return, by default None + + Raises + ------ + List[str] + the first top ids + """ + raise NotImplementedError("get_learnware_ids is not implemented") + + + def get_learnwares(self, top:int = None) -> List[Learnware]: + """get the list of learnwares + + Parameters + ---------- + top : int, optional + the first top element to return, by default None + + Raises + ------ + List[Learnware] + the first top learnwares + """ + raise NotImplementedError("get_learnwares is not implemented") class LearnwareSearcher: def __init__(self, organizer: LearnwareOrganizer = None): @@ -337,7 +252,14 @@ class LearnwareSearcher: def reset(self, organizer): self.learnware_oganizer = organizer - def __call__(self, user_info: BaseUserInfo): + def __call__(self, user_info: BaseUserInfo) + """Search learnwares based on user_info + + Parameters + ---------- + user_info : BaseUserInfo + user_info contains semantic_spec and stat_info + """ raise NotImplementedError("'__call__' method is not implemented in LearnwareSearcher") diff --git a/learnware/market/database_ops.py b/learnware/market/database_ops.py index f656425..c44aa01 100644 --- a/learnware/market/database_ops.py +++ b/learnware/market/database_ops.py @@ -117,25 +117,25 @@ class DatabaseOperations(object): pass pass - def update_learnware_semantic_spec(self, learnware_id: str, semantic_spec: dict): + def delete_learnware(self, id: str): 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.execute(text("DELETE FROM tb_learnware WHERE id=:id;"), dict(id=id)) conn.commit() pass pass - def delete_learnware(self, id: str): + def update_learnware_semantic_specification(self, id: str, semantic_spec: dict): with self.engine.connect() as conn: - conn.execute(text("DELETE FROM tb_learnware WHERE id=:id;"), dict(id=id)) + semantic_spec_str = json.dumps(semantic_spec) + r = conn.execute( + text("UPDATE tb_learnware SET semantic_spec=:semantic_spec WHERE id=:id;"), + dict(id=id, semantic_spec=semantic_spec_str), + ) conn.commit() pass pass - def update_learnware_semantic_specification(self, id: str, semantic_spec: dict): + def update_learnware_use_flag(self, id: str, semantic_spec: dict): with self.engine.connect() as conn: semantic_spec_str = json.dumps(semantic_spec) r = conn.execute( diff --git a/learnware/market/easy.py b/learnware/market/easy.py index c577a82..63a357e 100644 --- a/learnware/market/easy.py +++ b/learnware/market/easy.py @@ -949,11 +949,11 @@ class EasyMarket(LearnwareMarket): logger.warning("Learnware ID '%s' NOT Found!" % (ids)) return None - def update_learnware_semantic_spec(self, learnware_id: str, semantic_spec: dict) -> bool: + def update_learnware_semantic_specification(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) + self.dbops.update_learnware_semantic_specification(learnware_id=learnware_id, semantic_spec=semantic_spec) # update file folder_path = self.learnware_folder_list[learnware_id] diff --git a/learnware/market/easy/checker.py b/learnware/market/easy/checker.py index efb48b2..ce5d2b4 100644 --- a/learnware/market/easy/checker.py +++ b/learnware/market/easy/checker.py @@ -1,4 +1,6 @@ import traceback +import numpy as np +import torch from ..base import LearnwareChecker from ...logger import get_module_logger diff --git a/learnware/market/easy/database_ops.py b/learnware/market/easy/database_ops.py index 48ed0fb..61bc02d 100644 --- a/learnware/market/easy/database_ops.py +++ b/learnware/market/easy/database_ops.py @@ -1,10 +1,10 @@ from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import create_engine, text -from sqlalchemy import Column, Text, String +from sqlalchemy import Column, Integer, Text, DateTime, String import os import json -from ...learnware import get_learnware_from_dirpath -from ...logger import get_module_logger +from ..learnware import get_learnware_from_dirpath +from ..logger import get_module_logger logger = get_module_logger("database") DeclarativeBase = declarative_base() @@ -117,17 +117,6 @@ 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)) @@ -146,6 +135,16 @@ class DatabaseOperations(object): pass pass + def update_learnware_use_flag(self, id: str, use_flag: str): + with self.engine.connect() as conn: + r = conn.execute( + text("UPDATE tb_learnware SET use_flag=:use_flag WHERE id=:id;"), + dict(id=id, use_flag=use_flag), + ) + conn.commit() + pass + pass + def load_market(self): with self.engine.connect() as conn: cursor = conn.execute(text("SELECT id, semantic_spec, zip_path, folder_path, use_flag FROM tb_learnware;")) @@ -153,6 +152,7 @@ class DatabaseOperations(object): learnware_list = {} zip_list = {} folder_list = {} + use_flags = {} max_count = 0 for id, semantic_spec, zip_path, folder_path, use_flag in cursor: @@ -166,10 +166,11 @@ class DatabaseOperations(object): # assert new_learnware is not None zip_list[id] = zip_path folder_list[id] = folder_path + use_flags[id] = use_flag max_count = max(max_count, int(id)) pass - return learnware_list, zip_list, folder_list, max_count + 1 + return learnware_list, zip_list, folder_list, use_flags, max_count + 1 pass pass diff --git a/learnware/market/easy/organizer.py b/learnware/market/easy/organizer.py index d3a70f8..7619e8a 100644 --- a/learnware/market/easy/organizer.py +++ b/learnware/market/easy/organizer.py @@ -4,6 +4,7 @@ import copy import torch import zipfile import traceback +import tempfile import numpy as np import pandas as pd from rapidfuzz import fuzz @@ -11,8 +12,10 @@ from cvxopt import solvers, matrix from shutil import copyfile, rmtree from typing import Tuple, Any, List, Union, Dict +from .database_ops import DatabaseOperations +from .checker import EasyChecker from ..base import LearnwareMarket, BaseUserInfo -from ..database_ops import DatabaseOperations + from ... import utils from ...config import C as conf @@ -28,8 +31,11 @@ logger = get_module_logger("easy_organizer") class EasyOrganizer(LearnwareOrganizer): - def reset(self, market_id, rebuild=False): - self.market_id = market_id + def __init__(self, market_id, checker: 'EasyChecker' = None, rebuild: bool = False): + self.reset(market_id=market_id, checker=checker, rebuild=rebuild) + + def reset(self, market_id, checker: EasyChecker = None, rebuild: bool = False): + super(EasyOrganizer, self).reset(market_id=market_id, checker=checker) self.reload_market(rebuild=rebuild) def reload_market(self, rebuild=False) -> bool: @@ -48,6 +54,7 @@ class EasyOrganizer(LearnwareOrganizer): self.learnware_list = {} # id: Learnware self.learnware_zip_list = {} self.learnware_folder_list = {} + self.use_flags = {} self.count = 0 self.semantic_spec_list = conf.semantic_specs self.dbops = DatabaseOperations(conf.database_url, "market_" + self.market_id) @@ -63,10 +70,10 @@ class EasyOrganizer(LearnwareOrganizer): os.makedirs(self.learnware_pool_path, exist_ok=True) os.makedirs(self.learnware_zip_pool_path, exist_ok=True) os.makedirs(self.learnware_folder_pool_path, exist_ok=True) - self.learnware_list, self.learnware_zip_list, self.learnware_folder_list, self.count = self.dbops.load_market() + self.learnware_list, self.learnware_zip_list, self.learnware_folder_list, self.use_flags, self.count = self.dbops.load_market() - def add_learnware(self, zip_path: str, semantic_spec: dict, learnware_id: str = None, check: bool = False) -> Tuple[str, bool]: + def add_learnware(self, zip_path: str, semantic_spec: dict, id: str = None, check: bool = True) -> Tuple[str, bool]: """Add a learnware into the market. .. note:: @@ -92,24 +99,24 @@ class EasyOrganizer(LearnwareOrganizer): if not os.path.exists(zip_path): logger.warning("Zip Path NOT Found! Fail to add learnware.") - return None, self.INVALID_LEARNWARE + return None, EasyChecker.INVALID_LEARNWARE try: if len(semantic_spec["Data"]["Values"]) == 0: logger.warning("Illegal semantic specification, please choose Data.") - return None, self.INVALID_LEARNWARE + return None, EasyChecker.INVALID_LEARNWARE if len(semantic_spec["Task"]["Values"]) == 0: logger.warning("Illegal semantic specification, please choose Task.") - return None, self.INVALID_LEARNWARE + return None, EasyChecker.INVALID_LEARNWARE if len(semantic_spec["Library"]["Values"]) == 0: logger.warning("Illegal semantic specification, please choose Device.") - return None, self.INVALID_LEARNWARE + return None, EasyChecker.INVALID_LEARNWARE if len(semantic_spec["Name"]["Values"]) == 0: logger.warning("Illegal semantic specification, please provide Name.") - return None, self.INVALID_LEARNWARE + return None, EasyChecker.INVALID_LEARNWARE if len(semantic_spec["Description"]["Values"]) == 0 and len(semantic_spec["Scenario"]["Values"]) == 0: logger.warning("Illegal semantic specification, please provide Scenario or Description.") - return None, self.INVALID_LEARNWARE + return None, EasyChecker.INVALID_LEARNWARE if ( semantic_spec["Data"]["Type"] != "Class" or semantic_spec["Task"]["Type"] != "Class" @@ -119,17 +126,15 @@ class EasyOrganizer(LearnwareOrganizer): or semantic_spec["Description"]["Type"] != "String" ): logger.warning("Illegal semantic specification, please provide the right type.") - return None, self.INVALID_LEARNWARE + return None, EasyChecker.INVALID_LEARNWARE except: print(semantic_spec) logger.warning("Illegal semantic specification, some keys are missing.") - return None, self.INVALID_LEARNWARE + return None, EasyChecker.INVALID_LEARNWARE logger.info("Get new learnware from %s" % (zip_path)) - if learnware_id is not None: - id = learnware_id - else: - id = "%08d" % (self.count) + + id = id if id is not None else "%08d" % (self.count) target_zip_dir = os.path.join(self.learnware_zip_pool_path, "%s.zip" % (id)) target_folder_dir = os.path.join(self.learnware_folder_pool_path, id) copyfile(zip_path, target_zip_dir) @@ -148,137 +153,167 @@ class EasyOrganizer(LearnwareOrganizer): rmtree(target_folder_dir) except: pass - return None, self.INVALID_LEARNWARE + return None, EasyChecker.INVALID_LEARNWARE if new_learnware is None: - return None, self.INVALID_LEARNWARE + return None, EasyChecker.INVALID_LEARNWARE - if check and self.checker + learnwere_status = EasyChecker.USABLE_LEARWARE if check is False else self.checker.check_learnware(new_learnware) self.dbops.add_learnware( id=id, semantic_spec=semantic_spec, zip_path=target_zip_dir, folder_path=target_folder_dir, - use_flag=LearnwareChecker.USABLE_LEARWARE, + use_flag=learnwere_status, ) self.learnware_list[id] = new_learnware self.learnware_zip_list[id] = target_zip_dir self.learnware_folder_list[id] = target_folder_dir + self.use_flags[id] = learnwere_status self.count += 1 - return id, LearnwareChecker.USABLE_LEARWARE - - def add_learnware(self, zip_path: str, semantic_spec: dict) -> Tuple[str, bool]: - """Add a learnware into the market. - - .. note:: - - Given a prediction of a certain time, all signals before this time will be prepared well. + return id, learnwere_status + def delete_learnware(self, id: str) -> bool: + """Delete Learnware from market Parameters ---------- - zip_path : str - Filepath for learnware model, a zipped file. - semantic_spec : dict - semantic_spec for new learnware, in dictionary format. + id : str + Learnware to be deleted Returns ------- - Tuple[str, int] - - str indicating model_id - - int indicating what the flag of learnware is added. - + bool + True for successful operation. + False for id not found. """ - semantic_spec = copy.deepcopy(semantic_spec) - - if not os.path.exists(zip_path): - logger.warning("Zip Path NOT Found! Fail to add learnware.") - return None, self.INVALID_LEARNWARE - - try: - if len(semantic_spec["Data"]["Values"]) == 0: - logger.warning("Illegal semantic specification, please choose Data.") - return None, self.INVALID_LEARNWARE - if len(semantic_spec["Task"]["Values"]) == 0: - logger.warning("Illegal semantic specification, please choose Task.") - return None, self.INVALID_LEARNWARE - if len(semantic_spec["Library"]["Values"]) == 0: - logger.warning("Illegal semantic specification, please choose Device.") - return None, self.INVALID_LEARNWARE - if len(semantic_spec["Name"]["Values"]) == 0: - logger.warning("Illegal semantic specification, please provide Name.") - return None, self.INVALID_LEARNWARE - if len(semantic_spec["Description"]["Values"]) == 0 and len(semantic_spec["Scenario"]["Values"]) == 0: - logger.warning("Illegal semantic specification, please provide Scenario or Description.") - return None, self.INVALID_LEARNWARE - if ( - semantic_spec["Data"]["Type"] != "Class" - or semantic_spec["Task"]["Type"] != "Class" - or semantic_spec["Library"]["Type"] != "Class" - or semantic_spec["Scenario"]["Type"] != "Tag" - or semantic_spec["Name"]["Type"] != "String" - or semantic_spec["Description"]["Type"] != "String" - ): - logger.warning("Illegal semantic specification, please provide the right type.") - return None, self.INVALID_LEARNWARE - except: - logger.info(f"Semantic specification: {semantic_spec}") - logger.warning("Illegal semantic specification, some keys are missing.") - return None, self.INVALID_LEARNWARE - - logger.info("Get new learnware from %s" % (zip_path)) - id = "%08d" % (self.count) - target_zip_dir = os.path.join(self.learnware_zip_pool_path, "%s.zip" % (id)) - target_folder_dir = os.path.join(self.learnware_folder_pool_path, id) - copyfile(zip_path, target_zip_dir) - - with zipfile.ZipFile(target_zip_dir, "r") as z_file: - z_file.extractall(target_folder_dir) - logger.info("Learnware move to %s, and unzip to %s" % (target_zip_dir, target_folder_dir)) - - try: - new_learnware = get_learnware_from_dirpath( + if not id in self.learnware_list: + logger.warning("Learnware id:'{}' NOT Found!".format(id)) + return False + + zip_dir = self.learnware_zip_list[id] + os.remove(zip_dir) + folder_dir = self.learnware_folder_list[id] + rmtree(folder_dir) + self.learnware_list.pop(id) + self.learnware_zip_list.pop(id) + self.learnware_folder_list.pop(id) + self.use_flags.pop(id) + self.dbops.delete_learnware(id=id) + + return True + + def update_learnware( + self, id: str, zip_path: str = None, semantic_spec: dict = None, check: bool = True + ): + """TODO: update should pass the semantic check too + """ + assert zip_path is None and semantic_spec is None, f"at least one of 'zip_path' and 'semantic_spec' should not be None when update learnware" + + if semantic_spec is not None: + self.dbops.update_learnware_semantic_specification(id, semantic_spec) + else: + semantic_spec = self.learnware_list[id] + + if zip_path is not None: + target_zip_dir = self.learnware_zip_list[id] + target_folder_dir = self.learnware_folder_list[id] + + if check: + with tempfile.TemporaryDirectory(prefix="learnware_") as tempdir: + with zipfile.ZipFile(zip_path, "r") as z_file: + z_file.extractall(tempdir) + + try: + new_learnware = get_learnware_from_dirpath( + id=id, semantic_spec=semantic_spec, learnware_dirpath=tempdir + ) + except Exception: + return False, EasyChecker.INVALID_LEARNWARE + + if new_learnware is None: + return False, EasyChecker.INVALID_LEARNWARE + + learnwere_status = self.checker.check_learnware(new_learnware) + else: + learnwere_status = EasyChecker.USABLE_LEARWARE + + self.dbops.update_learnware_use_flag(id, learnwere_status) + copyfile(zip_path, target_zip_dir) + with zipfile.ZipFile(target_zip_dir, "r") as z_file: + z_file.extractall(target_folder_dir) + self.learnware_list[id] = get_learnware_from_dirpath( id=id, semantic_spec=semantic_spec, learnware_dirpath=target_folder_dir ) - except: - try: - os.remove(target_zip_dir) - rmtree(target_folder_dir) - except: - pass - return None, self.INVALID_LEARNWARE + + return True, learnwere_status + + else: + self.learnware_list[id].get_specification().update_semantic_spec(semantic_spec) + return self.use_flags[id] + + def get_learnware_by_ids(self, ids: Union[str, List[str]]) -> Union[Learnware, List[Learnware]]: + """Search learnware by id or list of ids. - if new_learnware is None: - return None, self.INVALID_LEARNWARE + Parameters + ---------- + ids : Union[str, List[str]] + Give a id or a list of ids + str: id of targer learware + List[str]: A list of ids of target learnwares - check_flag = self.check_learnware(new_learnware) + Returns + ------- + Union[Learnware, List[Learnware]] + Return target learnware or list of target learnwares. + None for Learnware NOT Found. + """ + if isinstance(ids, list): + ret = [] + for id in ids: + if id in self.learnware_list: + ret.append(self.learnware_list[id]) + else: + logger.warning("Learnware ID '%s' NOT Found!" % (id)) + ret.append(None) + return ret + else: + try: + return self.learnware_list[ids] + except: + logger.warning("Learnware ID '%s' NOT Found!" % (ids)) + return None - self.dbops.add_learnware( - id=id, - semantic_spec=semantic_spec, - zip_path=target_zip_dir, - folder_path=target_folder_dir, - use_flag=check_flag, - ) + def get_learnware_path_by_ids(self, ids: Union[str, List[str]]) -> Union[Learnware, List[Learnware]]: + """Get Zipped Learnware file by id - self.learnware_list[id] = new_learnware - self.learnware_zip_list[id] = target_zip_dir - self.learnware_folder_list[id] = target_folder_dir - self.count += 1 - return id, check_flag - + Parameters + ---------- + ids : Union[str, List[str]] + Give a id or a list of ids + str: id of targer learware + List[str]: A list of ids of target learnwares - def get_learnware_ids(self, top:int = None): - if top is None: - return list(self.learnware_list.keys()) - else: - return list(self.learnware_list.keys())[:top] - - - def get_learnwares(self, top:int = None): - if top is None: - return list(self.learnware_list.values()) + Returns + ------- + Union[Learnware, List[Learnware]] + Return the path for target learnware or list of path. + None for Learnware NOT Found. + """ + if isinstance(ids, list): + ret = [] + for id in ids: + if id in self.learnware_zip_list: + ret.append(self.learnware_zip_list[id]) + else: + logger.warning("Learnware ID '%s' NOT Found!" % (id)) + ret.append(None) + return ret else: - return list(self.learnware_list.values())[:top] \ No newline at end of file + try: + return self.learnware_zip_list[ids] + except: + logger.warning("Learnware ID '%s' NOT Found!" % (ids)) + return None \ No newline at end of file