beimingwu
/
learnware
Not watched
1
0
Fork 0
Code
Releases 1 Wiki evaluate Activity
Issues 0 Pull Requests 0 Datasets Model Cloudbrain HPC
Browse Source

[MNT] maintain use_flags in easy market

tags/v0.3.2
bxdd 2 years ago
parent
01c96c1abe
commit
50551d4265
6 changed files with 241 additions and 281 deletions
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +57
    -135
      learnware/market/base.py
  2. +9
    -9
      learnware/market/database_ops.py
  3. +2
    -2
      learnware/market/easy.py
  4. +2
    -0
      learnware/market/easy/checker.py
  5. +16
    -15
      learnware/market/easy/database_ops.py
  6. +155
    -120
      learnware/market/easy/organizer.py

+ 57
- 135
learnware/market/base.py View File

@@ -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")



+ 9
- 9
learnware/market/database_ops.py View File

@@ -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(


+ 2
- 2
learnware/market/easy.py View File

@@ -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]


+ 2
- 0
learnware/market/easy/checker.py View File

@@ -1,4 +1,6 @@
import traceback
import numpy as np
import torch

from ..base import LearnwareChecker
from ...logger import get_module_logger


+ 16
- 15
learnware/market/easy/database_ops.py View File

@@ -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

+ 155
- 120
learnware/market/easy/organizer.py View File

@@ -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]
try:
return self.learnware_zip_list[ids]
except:
logger.warning("Learnware ID '%s' NOT Found!" % (ids))
return None

Write Preview
Loading…
Cancel
Save