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

Merge branch 'main' of https://github.com/Learnware-LAMDA/Learnware into search_result

tags/v0.3.2
bxdd 2 years ago
parent
262cc06b28 c38d0ab7e7
commit
9f13198ddd
6 changed files with 289 additions and 220 deletions
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +6
    -5
      learnware/client/learnware_client.py
  2. +131
    -100
      learnware/tests/benchmarks/__init__.py
  3. +14
    -11
      learnware/tests/benchmarks/config.py
  4. +56
    -37
      tests/test_function/test_search.py
  5. +38
    -30
      tests/test_learnware_client/test_all_learnware.py
  6. +44
    -37
      tests/test_learnware_client/test_upload.py

+ 6
- 5
learnware/client/learnware_client.py View File

@@ -54,6 +54,7 @@ class SemanticSpecificationKey(Enum):
TASK_TYPE = "Task"
LIBRARY_TYPE = "Library"
SENARIOES = "Scenario"
LICENSE = "License"


class LearnwareClient:
@@ -69,14 +70,14 @@ class LearnwareClient:
self.tempdir_list = []
self.login_status = False
atexit.register(self.cleanup)
def is_connected(self):
url = f"{self.host}/auth/login_by_token"
response = requests.post(url)
if response.status_code == 404:
return False
return True
def login(self, email, token):
url = f"{self.host}/auth/login_by_token"

@@ -89,10 +90,10 @@ class LearnwareClient:
token = result["data"]["token"]
self.headers = {"Authorization": f"Bearer {token}"}
self.login_status = True
def is_login(self):
return self.login_status
@require_login
def logout(self):
url = f"{self.host}/auth/logout"
@@ -277,7 +278,7 @@ class LearnwareClient:
returns["multiple"]["learnware_ids"].append(multi_learnware["learnware_id"])
returns["multiple"]["semantic_specifications"].append(multi_learnware["semantic_specification"])
returns["multiple"]["matching"] = learnware["matching"]
# Delete temp json file
os.remove(temp_file_name)



+ 131
- 100
learnware/tests/benchmarks/__init__.py View File

@@ -1,130 +1,161 @@
import os
import pickle
import atexit
import tempfile
import zipfile
from dataclasses import dataclass, field
from typing import Optional, List, Union, Tuple
from dataclasses import dataclass
from typing import Tuple, Optional, List, Union

from .config import OnlineBenchmark, online_benchmarks
from .config import BenchmarkConfig, benchmark_configs
from ..data import GetData
from ...config import C


@dataclass
class Benchmark:
learnware_ids: List[str]
user_num: int
unlabeled_feature_paths: List[str]
unlabeled_groudtruths_paths: List[str]
labeled_feature_paths: Optional[List[str]] = None
labeled_label_paths: Optional[List[str]] = None
test_X_paths: List[str]
test_y_paths: List[str]
train_X_paths: Optional[List[str]] = None
train_y_paths: Optional[List[str]] = None
extra_info_path: Optional[str] = None
# TODO: add more method for benchmark
def get_unlabeled_data(self, user_ids: Union[str, List[str]]):

def get_test_data(self, user_ids: Union[str, List[str]]):
if isinstance(user_ids, str):
user_ids = [user_ids]
ret = []
for user_id in user_ids:
with open(self.unlabeled_feature_paths[user_id], "rb") as fin:
unlabeled_feature = pickle.load(fin)
with open(self.unlabeled_groudtruths_paths[user_id], "rb") as fin:
unlabeled_groudtruth = pickle.load(fin)
ret.append((unlabeled_feature, unlabeled_groudtruth))
with open(self.test_X_paths[user_id], "rb") as fin:
test_X = pickle.load(fin)
with open(self.test_y_paths[user_id], "rb") as fin:
test_y = pickle.load(fin)
ret.append((test_X, test_y))

return ret
def get_labeled_data(self, user_ids):
if self.labeled_feature_paths is None or self.labeled_label_paths is None:
def get_train_data(self, user_ids):
if self.train_X_paths is None or self.train_y_paths is None:
return None
if isinstance(user_ids, str):
user_ids = [user_ids]
ret = []
for user_id in user_ids:
with open(self.labeled_feature_paths[user_id], "rb") as fin:
labeled_feature = pickle.load(fin)
with open(self.labeled_label_paths[user_id], "rb") as fin:
labeled_groudtruth = pickle.load(fin)
ret.append((labeled_feature, labeled_groudtruth))
with open(self.train_X_paths[user_id], "rb") as fin:
train_X = pickle.load(fin)
with open(self.train_y_paths[user_id], "rb") as fin:
train_y = pickle.load(fin)
ret.append((train_X, train_y))

return ret

class LearnwareBenchmark:
def __init__(self):
self.online_benchmarks = online_benchmarks
self.tempdir_list = []
atexit.register(self.cleanup)
self.benchmark_configs = benchmark_configs

def list_benchmarks(self):
return list(self.online_benchmarks.keys())
def get_benchmark(self, online_benchmark: Union[str, OnlineBenchmark]):
if isinstance(online_benchmark, str):
online_benchmark = self.online_benchmarks[online_benchmark]
self.tempdir_list.append(tempfile.TemporaryDirectory(prefix="learnware_benchmark"))
save_folder = self.tempdir_list[-1].name
unlabeled_data_localpath = os.path.join(save_folder, "unlabeled_data.zip")
GetData().download_file(online_benchmark.unlabeled_data_path, unlabeled_data_localpath)
unlabeled_feature_paths = []
unlabeled_groudtruth_paths = []
with zipfile.ZipFile(unlabeled_data_localpath, "r") as z_file:
unlabeled_data_dirpath = os.path.join(save_folder, "unlabeled_data")
z_file.extractall(unlabeled_data_dirpath)
for user_id in range(online_benchmark.user_num):
user_feature_filepath = os.path.isfile(os.path.join(unlabeled_data_dirpath, f"user{user_id}_feature.pkl"))
user_groudtruth_filepath = os.path.isfile(os.path.join(unlabeled_data_dirpath, f"user{user_id}_groudtruth.pkl"))
assert os.path.isfile(user_feature_filepath), f"user {user_id} unlabeled feature is not valid!"
assert os.path.isfile(user_groudtruth_filepath), f"user {user_id} unlabeled groudtruth is not valid!"
unlabeled_feature_paths.append(user_feature_filepath)
unlabeled_groudtruth_paths.append(user_groudtruth_filepath)

labeled_feature_paths = None
labeled_label_paths = None
if online_benchmark.labeled_data_path is not None:
labeled_data_localpath = os.path.join(save_folder, "labeled_data.zip")
GetData().download_file(online_benchmark.labeled_data_path, labeled_data_localpath)
labeled_feature_paths = []
labeled_label_paths = []

with zipfile.ZipFile(labeled_data_localpath, "r") as z_file:
labeled_data_dirpath = os.path.join(save_folder, "labeled_data")
z_file.extractall(labeled_data_dirpath)
for user_id in range(online_benchmark.user_num):
user_feature_filepath = os.path.isfile(os.path.join(labeled_data_dirpath, f"user{user_id}_feature.pkl"))
user_groudtruth_filepath = os.path.isfile(os.path.join(labeled_data_dirpath, f"user{user_id}_label.pkl"))
assert os.path.isfile(user_feature_filepath), f"user {user_id} labeled feature is not valid!"
assert os.path.isfile(user_groudtruth_filepath), f"user {user_id} labeled label is not valid!"
labeled_feature_paths.append(user_feature_filepath)
labeled_label_paths.append(user_groudtruth_filepath)
extra_zip_localpath = None
if online_benchmark.extra_info_path is not None:
extra_zip_localpath = os.path.join(save_folder, os.path.basename(online_benchmark.extra_info_path))
GetData().download_file(online_benchmark.extra_info_path, extra_zip_localpath)
return list(self.benchmark_configs.keys())

def _check_cache_data_valid(self, benchmark_config: BenchmarkConfig, data_type: str) -> bool:
"""Check if the cache data is valid

Parameters
----------
benchmark_config : BenchmarkConfig
benchmark config
data_type : str
"test" for test data or "train" for train data

Returns
-------
bool
A flag indicating if the cache data is valid
"""
cache_folder = os.path.join(C.cache_path, benchmark_config.name, f"{data_type}_data")
if os.path.exists(cache_folder):
for user_id in range(benchmark_config.user_num):
X_path = os.path.join(cache_folder, f"user{user_id}_X.pkl")
y_path = os.path.join(cache_folder, f"user{user_id}_X.pkl")
if not os.path.isfile(X_path) or not os.path.isfile(y_path):
return False
return True
else:
return False

def _download_data(self, download_path: str, save_path: str):
"""Download data from backend

Parameters
----------
download_path : str
data path for download in backend
save_path : str
local cache path for saving data
"""
with tempfile.TemporaryDirectory(prefix="learnware_benchmark_") as tempdir:
test_data_zippath = os.path.join(tempdir, "benchmark_data.zip")
GetData().download_file(download_path, test_data_zippath)

os.makedirs(save_path, exist_ok=True)
with zipfile.ZipFile(test_data_zippath, "r") as z_file:
z_file.extractall(save_path)

def _load_cache_data(self, benchmark_config: BenchmarkConfig, data_type: str) -> Tuple(List[str], List[str]):
"""Load data from local cache path

Parameters
----------
benchmark_config : BenchmarkConfig
benchmark config
data_type : str
"test" for test data or "train" for train data
"""
cache_folder = os.path.join(C.cache_path, benchmark_config.name, f"{data_type}_data")
if not self._check_cache_data_valid(benchmark_config, data_type):
download_path = getattr(benchmark_config, f"{data_type}_data_path", None)
self._download_data(download_path, cache_folder)

X_paths, y_paths = [], []
for user_id in range(benchmark_config.user_num):
user_X_path = os.path.join(cache_folder, f"user{user_id}_X.pkl")
user_y_path = os.path.join(cache_folder, f"user{user_id}_y.pkl")
assert os.path.isfile(user_X_path), f"user {user_id} {data_type}_X is not valid!"
assert os.path.isfile(user_y_path), f"user {user_id} {data_type}_y is not valid!"
X_paths.append(user_X_path)
y_paths.append(user_y_path)

def get_benchmark(self, benchmark_config: Union[str, BenchmarkConfig]):
if isinstance(benchmark_config, str):
benchmark_config = self.benchmark_configs[benchmark_config]

# Load test data
test_X_paths, test_y_paths = self._load_cache_data(benchmark_config, "test")

# Load train data
train_X_paths, train_y_paths = None, None
if benchmark_config.train_data_path is not None:
train_X_paths, train_y_paths = self._load_cache_data(benchmark_config, "train")

# Load extra info
extra_info_path = None
if benchmark_config.extra_info_path is not None:
extra_info_path = os.path.join(C.cache_path, benchmark_config.name, "extra_info")
if not os.path.exists(extra_info_path):
self._download_data(benchmark_config.extra_info_path, extra_info_path)

return Benchmark(
learnware_ids=online_benchmark.learnware_ids,
user_num=online_benchmark.user_num,
unlabeled_feature_paths=unlabeled_feature_paths,
unlabeled_groudtruths_paths=unlabeled_groudtruth_paths,
labeled_feature_paths=labeled_feature_paths,
labeled_label_paths=labeled_label_paths,
extra_info_path=extra_zip_localpath,
learnware_ids=benchmark_config.learnware_ids,
user_num=benchmark_config.user_num,
test_X_paths=test_X_paths,
test_y_paths=test_y_paths,
train_X_paths=train_X_paths,
train_y_paths=train_y_paths,
extra_info_path=extra_info_path,
)
def cleanup(self):
for tempdir in self.tempdir_list:
tempdir.cleanup()

+ 14
- 11
learnware/tests/benchmarks/config.py View File

@@ -1,21 +1,24 @@
from dataclasses import dataclass, field
from dataclasses import dataclass
from typing import Optional, List
from ...learnware import Learnware

@dataclass
class OnlineBenchmark:
class BenchmarkConfig:
name: str
learnware_ids: List[str]
user_num: int
unlabeled_data_path: str
labeled_data_path: Optional[str] = None
test_data_path: str
train_data_path: Optional[str] = None
extra_info_path: Optional[str] = None

online_benchmarks = {
"example": OnlineBenchmark(

benchmark_configs = {
"example": BenchmarkConfig(
name="example",
learnware_ids=["00001951", "00001980", "00001987"],
user_num=3,
unlabeled_data_path="example_path1",
labeled_data_path="example_path2",
extra_info_path="example_path3"
test_data_path="example_path1",
train_data_path="example_path2",
extra_info_path="example_path3",
)
}
}

+ 56
- 37
tests/test_function/test_search.py View File

@@ -4,6 +4,7 @@ import tempfile
import logging

import learnware

learnware.init(logging_level=logging.WARNING)

from learnware.learnware import Learnware
@@ -11,9 +12,10 @@ from learnware.client import LearnwareClient
from learnware.market import instantiate_learnware_market, BaseUserInfo, EasySemanticChecker
from learnware.config import C


class TestSearch(unittest.TestCase):
client = LearnwareClient()
@classmethod
def setUpClass(cls):
cls.market = instantiate_learnware_market(market_id="search_test", name="hetero", rebuild=True)
@@ -31,46 +33,62 @@ class TestSearch(unittest.TestCase):
learnware_zippath = os.path.join(tempdir, f"learnware_{learnware_id}.zip")
try:
cls.client.download_learnware(learnware_id=learnware_id, save_path=learnware_zippath)
semantic_spec = cls.client.load_learnware(learnware_path=learnware_zippath).get_specification().get_semantic_spec()
semantic_spec = (
cls.client.load_learnware(learnware_path=learnware_zippath)
.get_specification()
.get_semantic_spec()
)
except Exception:
print("'learnware_id' is passed due to the network problem.")
cls.market.add_learnware(learnware_zippath, learnware_id=learnware_id, semantic_spec=semantic_spec, checker_names=["EasySemanticChecker"])
@unittest.skipIf(not client.is_connected(), "Client can not connect!")
cls.market.add_learnware(
learnware_zippath,
learnware_id=learnware_id,
semantic_spec=semantic_spec,
checker_names=["EasySemanticChecker"],
)

def _skip_test(self):
if not self.client.is_connected():
print("Client can not connect!")
return True
return False

def test_image_search(self):
learnware_id = "00000619"
try:
learnware: Learnware = self.client.load_learnware(learnware_id=learnware_id)
except Exception:
print("'test_image_search' is passed due to the network problem.")
user_info = BaseUserInfo(stat_info=learnware.get_specification().get_stat_spec())
search_result = self.market.search_learnware(user_info)
print("Single Search Results:", search_result.get_single_results())
print("Multiple Search Results:", search_result.get_multiple_results())
@unittest.skipIf(not client.is_connected(), "Client can not connect!")
if not self._skip_test():
learnware_id = "00000619"
try:
learnware: Learnware = self.client.load_learnware(learnware_id=learnware_id)
except Exception:
print("'test_image_search' is passed due to the network problem.")
user_info = BaseUserInfo(stat_info=learnware.get_specification().get_stat_spec())
search_result = self.market.search_learnware(user_info)
print("Single Search Results:", search_result.get_single_results())
print("Multiple Search Results:", search_result.get_multiple_results())
def test_text_search(self):
learnware_id = "00000653"
try:
learnware: Learnware = self.client.load_learnware(learnware_id=learnware_id)
except Exception:
print("'test_text_search' is passed due to the network problem.")
user_info = BaseUserInfo(stat_info=learnware.get_specification().get_stat_spec())
search_result = self.market.search_learnware(user_info)
print("Single Search Results:", search_result.get_single_results())
print("Multiple Search Results:", search_result.get_multiple_results())
@unittest.skipIf(not client.is_connected(), "Client can not connect!")
if not self._skip_test():
learnware_id = "00000653"
try:
learnware: Learnware = self.client.load_learnware(learnware_id=learnware_id)
except Exception:
print("'test_text_search' is passed due to the network problem.")
user_info = BaseUserInfo(stat_info=learnware.get_specification().get_stat_spec())
search_result = self.market.search_learnware(user_info)
print("Single Search Results:", search_result.get_single_results())
print("Multiple Search Results:", search_result.get_multiple_results())
def test_table_search(self):
learnware_id = "00001950"
try:
learnware: Learnware = self.client.load_learnware(learnware_id=learnware_id)
except Exception:
print("'test_table_search' is passed due to the network problem.")
user_info = BaseUserInfo(stat_info=learnware.get_specification().get_stat_spec())
search_result = self.market.search_learnware(user_info)
print("Single Search Results:", search_result.get_single_results())
print("Multiple Search Results:", search_result.get_multiple_results())
if not self._skip_test():
learnware_id = "00001950"
try:
learnware: Learnware = self.client.load_learnware(learnware_id=learnware_id)
except Exception:
print("'test_table_search' is passed due to the network problem.")
user_info = BaseUserInfo(stat_info=learnware.get_specification().get_stat_spec())
search_result = self.market.search_learnware(user_info)
print("Single Search Results:", search_result.get_single_results())
print("Multiple Search Results:", search_result.get_multiple_results())


def suite():
_suite = unittest.TestSuite()
@@ -79,6 +97,7 @@ def suite():
_suite.addTest(TestSearch("test_table_search"))
return _suite


if __name__ == "__main__":
runner = unittest.TextTestRunner()
runner.run(suite())
runner.run(suite())

+ 38
- 30
tests/test_learnware_client/test_all_learnware.py View File

@@ -9,13 +9,14 @@ from learnware.client import LearnwareClient
from learnware.specification import generate_semantic_spec
from learnware.market import BaseUserInfo


class TestAllLearnware(unittest.TestCase):
client = LearnwareClient()
@classmethod
def setUpClass(cls) -> None:
config_path = os.path.join(os.path.dirname(__file__), "config.json")
if not os.path.exists(config_path):
data = {"email": None, "token": None}
with open(config_path, "w") as file:
@@ -25,40 +26,46 @@ class TestAllLearnware(unittest.TestCase):
data = json.load(file)
email = data.get("email")
token = data.get("token")
if email is None or token is None:
print("Please set email and token in config.json.")
else:
cls.client.login(email, token)

@unittest.skipIf(not client.is_login(), "Client doest not login!")
def _skip_test(self):
if not self.client.is_login():
print("Client does not login!")
return True
return False

def test_all_learnware(self):
max_learnware_num = 2000
semantic_spec = generate_semantic_spec()
user_info = BaseUserInfo(semantic_spec=semantic_spec, stat_info={})
result = self.client.search_learnware(user_info, page_size=max_learnware_num)
learnware_ids = result["single"]["learnware_ids"]
keys = [key for key in result["single"]["semantic_specifications"][0]]
print(f"result size: {len(learnware_ids)}")
print(f"key in result: {keys}")

failed_ids = []
with tempfile.TemporaryDirectory(prefix="learnware_") as tempdir:
for idx in learnware_ids:
zip_path = os.path.join(tempdir, f"test_{idx}.zip")
self.client.download_learnware(idx, zip_path)
with zipfile.ZipFile(zip_path, "r") as zip_file:
with zip_file.open("semantic_specification.json") as json_file:
semantic_spec = json.load(json_file)
try:
LearnwareClient.check_learnware(zip_path, semantic_spec)
print(f"check learnware {idx} succeed")
except:
failed_ids.append(idx)
print(f"check learnware {idx} failed!!!")

print(f"The currently failed learnware ids: {failed_ids}")
if not self._skip_test():
max_learnware_num = 2000
semantic_spec = generate_semantic_spec()
user_info = BaseUserInfo(semantic_spec=semantic_spec, stat_info={})
result = self.client.search_learnware(user_info, page_size=max_learnware_num)

learnware_ids = result["single"]["learnware_ids"]
keys = [key for key in result["single"]["semantic_specifications"][0]]
print(f"result size: {len(learnware_ids)}")
print(f"key in result: {keys}")

failed_ids = []
with tempfile.TemporaryDirectory(prefix="learnware_") as tempdir:
for idx in learnware_ids:
zip_path = os.path.join(tempdir, f"test_{idx}.zip")
self.client.download_learnware(idx, zip_path)
with zipfile.ZipFile(zip_path, "r") as zip_file:
with zip_file.open("semantic_specification.json") as json_file:
semantic_spec = json.load(json_file)
try:
LearnwareClient.check_learnware(zip_path, semantic_spec)
print(f"check learnware {idx} succeed")
except:
failed_ids.append(idx)
print(f"check learnware {idx} failed!!!")

print(f"The currently failed learnware ids: {failed_ids}")


def suite():
@@ -66,6 +73,7 @@ def suite():
_suite.addTest(TestAllLearnware("test_all_learnware"))
return _suite


if __name__ == "__main__":
runner = unittest.TextTestRunner()
runner.run(suite())

+ 44
- 37
tests/test_learnware_client/test_upload.py View File

@@ -6,13 +6,14 @@ import tempfile
from learnware.client import LearnwareClient
from learnware.specification import generate_semantic_spec


class TestUpload(unittest.TestCase):
client = LearnwareClient()
@classmethod
def setUpClass(cls) -> None:
config_path = os.path.join(os.path.dirname(__file__), "config.json")
if not os.path.exists(config_path):
data = {"email": None, "token": None}
with open(config_path, "w") as file:
@@ -22,50 +23,55 @@ class TestUpload(unittest.TestCase):
data = json.load(file)
email = data.get("email")
token = data.get("token")
if email is None or token is None:
print("Please set email and token in config.json.")
else:
cls.client.login(email, token)

@unittest.skipIf(not client.is_login(), "Client doest not login!")
def _skip_test(self):
if not self.client.is_login():
print("Client does not login!")
return True
return False

def test_upload(self):
input_description = {
"Dimension": 13,
"Description": {"0": "age", "1": "weight", "2": "body length", "3": "animal type", "4": "claw length"},
}
output_description = {
"Dimension": 1,
"Description": {
"0": "the probability of being a cat",
},
}
semantic_spec = generate_semantic_spec(
name="learnware_example",
description="Just a example for uploading a learnware",
data_type="Table",
task_type="Classification",
library_type="Scikit-learn",
scenarios=["Business", "Financial"],
input_description=input_description,
output_description=output_description,
)
assert isinstance(semantic_spec, dict)

download_learnware_id = "00000084"
with tempfile.TemporaryDirectory(prefix="learnware_") as tempdir:
zip_path = os.path.join(tempdir, f"test.zip")
self.client.download_learnware(download_learnware_id, zip_path)
learnware_id = self.client.upload_learnware(
learnware_zip_path=zip_path, semantic_specification=semantic_spec
if not self._skip_test():
input_description = {
"Dimension": 13,
"Description": {"0": "age", "1": "weight", "2": "body length", "3": "animal type", "4": "claw length"},
}
output_description = {
"Dimension": 2,
"Description": {"0": "cat", "1": "not cat"},
}
semantic_spec = generate_semantic_spec(
name="learnware_example",
description="Just a example for uploading a learnware",
data_type="Table",
task_type="Classification",
library_type="Scikit-learn",
scenarios=["Business", "Financial"],
license="MIT",
input_description=input_description,
output_description=output_description,
)
assert isinstance(semantic_spec, dict)

download_learnware_id = "00000084"
with tempfile.TemporaryDirectory(prefix="learnware_") as tempdir:
zip_path = os.path.join(tempdir, f"test.zip")
self.client.download_learnware(download_learnware_id, zip_path)
learnware_id = self.client.upload_learnware(
learnware_zip_path=zip_path, semantic_specification=semantic_spec
)

uploaded_ids = [learnware["learnware_id"] for learnware in self.client.list_learnware()]
assert learnware_id in uploaded_ids
uploaded_ids = [learnware["learnware_id"] for learnware in self.client.list_learnware()]
assert learnware_id in uploaded_ids

self.client.delete_learnware(learnware_id)
uploaded_ids = [learnware["learnware_id"] for learnware in self.client.list_learnware()]
assert learnware_id not in uploaded_ids
self.client.delete_learnware(learnware_id)
uploaded_ids = [learnware["learnware_id"] for learnware in self.client.list_learnware()]
assert learnware_id not in uploaded_ids


def suite():
@@ -73,6 +79,7 @@ def suite():
_suite.addTest(TestUpload("test_upload"))
return _suite


if __name__ == "__main__":
runner = unittest.TextTestRunner()
runner.run(suite())

Write Preview
Loading…
Cancel
Save