[MNT] enhance HeteroSearcher

learnware/market/__init__.py

@@ -3,7 +3,7 @@ from .base import BaseUserInfo, LearnwareMarket, BaseChecker, BaseOrganizer, Bas
 from .evolve_anchor import EvolvedAnchoredOrganizer
 from .evolve import EvolvedOrganizer
 from .easy2 import EasyOrganizer, EasySearcher, EasySemanticChecker, EasyStatChecker
-from .hetergeneous import HeteroMapTableOrganizer, HeteroMapTableSearcher
+from .hetergeneous import HeteroMapTableOrganizer, HeteroSearcher
 
 from .easy import EasyMarket
 from .classes import CondaChecker

learnware/market/hetergeneous/__init__.py

@@ -1,2 +1,2 @@
 from .organizer import HeteroMapTableOrganizer
-from .searcher import HeteroMapTableSearcher
+from .searcher import HeteroSearcher

learnware/market/hetergeneous/organizer/__init__.py

@@ -167,7 +167,6 @@ class HeteroMapTableOrganizer(EasyOrganizer):
             for learnware in learnware_list:
                 hetero_spec_path = os.path.join(hetero_mappings_save_path, f"{learnware.id}.npy")
                 self._update_learnware_specification(learnware, save_path=hetero_spec_path)
-                logger.info(f"Learnware {learnware.id} HeteroSpecification Successfully Saved")
         except Exception as err:
             logger.warning(f"Update learnware HeteroSpecification failed! Due to {err}")
 

learnware/market/hetergeneous/searcher.py

@@ -1,29 +1,122 @@
-from typing import List
+from typing import Tuple, List, Union
 
 from ...learnware import Learnware
 from ...logger import get_module_logger
+from ...specification import HeteroSpecification
 from ..base import BaseSearcher, BaseUserInfo
+from ..easy2 import EasySearcher
+from ..utils import parse_specification_type
 from .organizer import HeteroMapTableOrganizer
 
 logger = get_module_logger("hetero_searcher")
 
 
-class HeteroMapTableSearcher(BaseSearcher):
-    def __init__(self, organizer: HeteroMapTableOrganizer = None):
-        super(HeteroMapTableSearcher, self).__init__(organizer)
+class HeteroMapTableSearcher(EasySearcher):
+    def _convert_dist_to_score(
+        self, dist_list: List[float], dist_epsilon: float = 0.01, min_score: float = 0.92
+    ) -> List[float]:
+        if len(dist_list) == 0:
+            return []
+
+        min_dist, max_dist = min(dist_list), max(dist_list)
+        if min_dist == max_dist:
+            return [1 for dist in dist_list]
+        else:
+            max_score = (max_dist - min_dist) / (max_dist - dist_epsilon)
+
+            if min_dist < dist_epsilon:
+                dist_epsilon = min_dist
+            elif max_score < min_score:
+                dist_epsilon = max_dist - (max_dist - min_dist) / min_score
+
+            return [(max_dist - dist) / (max_dist - dist_epsilon) for dist in dist_list]
+
+    def _search_by_hetero_spec_single(
+        self, 
+        learnware_list: List[Learnware], 
+        user_hetero_spec: HeteroSpecification
+    ) -> Tuple[List[float], List[Learnware]]:
+        hetero_spec_list = [learnware.specification.get_stat_spec_by_name("HeteroSpecification") for learnware in learnware_list]
+        mmd_dist_list = []
+        for hetero_spec in hetero_spec_list:
+            mmd_dist = hetero_spec.dist(user_hetero_spec)
+            mmd_dist_list.append(mmd_dist)
+        
+        sorted_idx_list = sorted(range(len(learnware_list)), key=lambda k: mmd_dist_list[k])
+        sorted_dist_list = [mmd_dist_list[idx] for idx in sorted_idx_list]
+        sorted_learnware_list = [learnware_list[idx] for idx in sorted_idx_list]
 
-    def __call__(self, user_info: BaseUserInfo, check_status: int = None) -> Learnware:
+        return sorted_dist_list, sorted_learnware_list
+    
+    def _filter_by_hetero_spec_single(
+        self,
+        sorted_score_list: List[float],
+        learnware_list: List[Learnware],
+        filter_score: float = 0.5,
+        min_num: int = 5
+    ) -> Tuple[List[float], List[Learnware]]:
+        idx = min(min_num, len(learnware_list))
+        while idx < len(learnware_list):
+            if sorted_score_list[idx] < filter_score:
+                break
+            idx += 1
+        return sorted_score_list[:idx], learnware_list[:idx]
+
+
+    def __call__(
+        self, 
+        learnware_list: List[Learnware], 
+        user_info: BaseUserInfo, 
+    ) -> Tuple[List[float], List[Learnware], float, List[Learnware]]:
         # todo: use specially assigned search_gamma for calculating mmd dist
-        learnware_list = self.learnware_oganizer.get_learnwares()
-        target_learnware, min_dist = None, None
         user_hetero_spec = self.learnware_oganizer.generate_hetero_map_spec(user_info)
-        for learnware in learnware_list:
-            learnware_hetero_spec = learnware.specification.get_stat_spec_by_name("HeteroSpecification")
-            mmd_dist = learnware_hetero_spec.dist(user_hetero_spec)
-            if target_learnware is None or mmd_dist < min_dist:
-                min_dist = mmd_dist
-                target_learnware = learnware
-        return target_learnware
+        logger.info(f"After semantic search, learnware_list length is {len(learnware_list)}")
+
+        sorted_dist_list, single_learnware_list = self._search_by_hetero_spec_single(learnware_list, user_hetero_spec)
+        sorted_score_list = self._convert_dist_to_score(sorted_dist_list)
+
+        logger.info(f"After search by hetero spec, learnware_list length is {len(single_learnware_list)}")
+        sorted_score_list, single_learnware_list = self._filter_by_hetero_spec_single(
+            sorted_score_list, single_learnware_list
+        )
+
+        logger.info(f"After filter by hetero spec, learnware_list length is {len(single_learnware_list)}")
+        return sorted_score_list, single_learnware_list, None, None
+
+        
+        # for learnware in learnware_list:
+        #     learnware_hetero_spec = learnware.specification.get_stat_spec_by_name("HeteroSpecification")
+        #     mmd_dist = learnware_hetero_spec.dist(user_hetero_spec)
+        #     if target_learnware is None or mmd_dist < min_dist:
+        #         min_dist = mmd_dist
+        #         target_learnware = learnware
+        # return target_learnware
 
     def reset(self, organizer):
         self.learnware_oganizer = organizer
+
+class HeteroSearcher(EasySearcher):
+    def __init__(self, organizer: HeteroMapTableOrganizer = None):
+        super(HeteroSearcher, self).__init__(organizer)
+        self.hetero_stat_searcher = HeteroMapTableSearcher(organizer)
+
+    def reset(self, organizer):
+        super().reset(organizer)
+        self.hetero_stat_searcher.reset(organizer)
+
+    def __call__(
+        self, user_info: BaseUserInfo, check_status: int = None, max_search_num: int = 5, search_method: str = "greedy"
+    ) -> Tuple[List[float], List[Learnware], float, List[Learnware]]:
+        learnware_list = self.learnware_organizer.get_learnwares(check_status=check_status)
+        learnware_list = self.semantic_searcher(learnware_list, user_info)
+
+        if len(learnware_list) == 0:
+            return [], [], 0.0, []
+
+        if parse_specification_type(stat_specs=user_info.stat_info) is not None:
+            if user_info.semantic_spec["Input"]["Description"] is not None:
+                return self.hetero_stat_searcher(learnware_list, user_info)
+            else:
+                return self.stat_searcher(learnware_list, user_info, max_search_num, search_method)
+        else:
+            return None, learnware_list, 0.0, None

learnware/market/module.py

@@ -1,6 +1,6 @@
 from .base import LearnwareMarket
 from .easy2 import EasyOrganizer, EasySearcher, EasySemanticChecker, EasyStatChecker
-from .hetergeneous import HeteroMapTableOrganizer, HeteroMapTableSearcher
+from .hetergeneous import HeteroMapTableOrganizer, HeteroSearcher
 
 MARKET_CONFIG = {
     "easy": {
@@ -10,7 +10,7 @@ MARKET_CONFIG = {
     }, 
     "hetero": {
         "organizer": HeteroMapTableOrganizer(),
-        "searcher": HeteroMapTableSearcher(),
+        "searcher": HeteroSearcher(),
         "checker_list": [EasySemanticChecker(), EasyStatChecker()]
     }
 }