code review for open source

Signed-off-by: SiqiLuo <1587295470@qq.com>

examples/lifelong_learning/robot_dog_delivery/RFNet/evaluate.py

@@ -7,6 +7,10 @@ os.environ["KB_SERVER"] = "http://0.0.0.0:9020"
 os.environ["operator"] = "<"
 os.environ["model_threshold"] = "0.01"
 
+os.environ["S3_ENDPOINT_URL"] = "https://obs.cn-north-1.myhuaweicloud.com"
+os.environ["SECRET_ACCESS_KEY"] = "OYPxi4uD9k5E90z0Od3Ug99symbJZ0AfyB4oveQc"
+os.environ["ACCESS_KEY_ID"] = "EMPTKHQUGPO2CDUFD2YR"
+
 from sedna.core.lifelong_learning import LifelongLearning
 from sedna.datasources import TxtDataParse
 from sedna.common.config import Context

examples/lifelong_learning/robot_dog_delivery/RFNet/predict.py

@@ -11,11 +11,14 @@ os.environ["robo_skill"] = "ramp_detection"
 os.environ["ramp_detection"] = "s3://kubeedge/sedna-robo/models/ramp_train1_200.pth"
 os.environ["curb_detection"] = "s3://kubeedge/sedna-robo/models/2048x1024_80.pth"
 
+os.environ["S3_ENDPOINT_URL"] = "https://obs.cn-north-1.myhuaweicloud.com"
+os.environ["SECRET_ACCESS_KEY"] = "OYPxi4uD9k5E90z0Od3Ug99symbJZ0AfyB4oveQc"
+os.environ["ACCESS_KEY_ID"] = "EMPTKHQUGPO2CDUFD2YR"
+
 from sedna.datasources import BaseDataSource
 from sedna.core.lifelong_learning import LifelongLearning
-from sedna.common.config import Context
 
-from basemodel import Model, preprocess_frames
+from basemodel import Model
 
 
 def preprocess(samples):
@@ -23,7 +26,6 @@ def preprocess(samples):
     data.x = [samples]
     return data
 
-
 def postprocess(samples):
     image_names, imgs = [], []
     for sample in samples:
@@ -33,11 +35,8 @@ def postprocess(samples):
 
     return image_names, imgs
 
-
 def init_ll_job():
-    robo_skill = Context.get_parameters("robo_skill", "ramp_detection")
     estimator = Model(num_class=31,
-                      weight_path=Context.get_parameters(robo_skill),
                       save_predicted_image=True,
                       merge=True)
 
@@ -47,17 +46,7 @@ def init_ll_job():
     unseen_task_allocation = {
         "method": "UnseenTaskAllocationDefault"
     }
-    unseen_sample_recognition = {
-        "method": "OodIdentification",
-        "param": {
-            "OOD_thresh": float(Context.get_parameters("OOD_thresh")),
-            "OOD_model": Context.get_parameters("OOD_model"),
-            "OOD_backup_model": Context.get_parameters(robo_skill),
-            "preprocess_func": preprocess_frames,
-            "base_model": Model
-        }
-    }
-
+   
     ll_job = LifelongLearning(
         estimator,
         unseen_estimator=unseen_task_processing,
@@ -68,7 +57,7 @@ def init_ll_job():
         inference_integrate=None,
         task_update_decision=None,
         unseen_task_allocation=unseen_task_allocation,
-        unseen_sample_recognition=unseen_sample_recognition,
+        unseen_sample_recognition=None,
         unseen_sample_re_recognition=None)
     return ll_job
 

examples/lifelong_learning/robot_dog_delivery/RFNet/robo_infer_service.py

@@ -173,7 +173,6 @@ class InferenceServer(BaseServer):  # pylint: disable=too-many-arguments
 
         self.index_frame = self.index_frame + 1
 
-        # img_rgb = cv2.resize(np.array(self.image), (2048, 1024), interpolation=cv2.INTER_CUBIC)
         img_rgb = Image.fromarray(np.array(self.image))
         if depth:
             depth_contents = await depth.read()

examples/lifelong_learning/robot_dog_delivery/RFNet/test.py

@@ -1,5 +0,0 @@
-
-with open("/home/lsq/RFNet/data_index/train.txt", "r") as f:
-    test_data = f.readlines()
-
-print()

examples/lifelong_learning/robot_dog_delivery/RFNet/train.py

@@ -5,6 +5,10 @@ os.environ["OUTPUT_URL"] = "s3://kubeedge/sedna-robo/kb/"
 os.environ["HAS_COMPLETED_INITIAL_TRAINING"] = "false"
 os.environ["KB_SERVER"] = "http://0.0.0.0:9020"
 
+os.environ["S3_ENDPOINT_URL"] = "https://obs.cn-north-1.myhuaweicloud.com"
+os.environ["SECRET_ACCESS_KEY"] = "OYPxi4uD9k5E90z0Od3Ug99symbJZ0AfyB4oveQc"
+os.environ["ACCESS_KEY_ID"] = "EMPTKHQUGPO2CDUFD2YR"
+
 from sedna.core.lifelong_learning import LifelongLearning
 from sedna.common.config import Context, BaseConfig
 from sedna.datasources import TxtDataParse

lib/sedna/algorithms/seen_task_learning/task_definition/task_definition_simple.py

@@ -8,7 +8,7 @@ from ..artifact import Task
 @ClassFactory.register(ClassType.STP)
 class TaskDefinitionSimple:
     """
-    Dividing datasets based on the their origins.
+    Dividing datasets based on the their original sites.
 
     Parameters
     ----------

lib/sedna/algorithms/unseen_task_detection/unseen_sample_recognition/__init__.py

@@ -13,6 +13,4 @@
 # limitations under the License.
 
 from . import unseen_sample_recognition
-from . import unseen_sample_recognition_robotic
-from . import ood_identification
 from . import unseen_sample_detection

lib/sedna/algorithms/unseen_task_detection/unseen_sample_recognition/ood_identification.py

@@ -1,219 +0,0 @@
-from typing import Tuple, List
-
-import torch
-import numpy as np
-from torch.utils.data import DataLoader
-import pandas as pd
-from sedna.common.constant import KBResourceConstant
-from sedna.backend import set_backend
-from sedna.common.file_ops import FileOps
-from sedna.datasources import BaseDataSource
-from sedna.common.class_factory import ClassFactory, ClassType
-from sedna.algorithms.seen_task_learning.artifact import Model, Task
-
-__all__ = ('OodIdentification',)
-
-
-@ClassFactory.register(ClassType.UTD, alias="OodIdentification")
-class OodIdentification:
-    """
-    Corresponding to `OodIdentification`
-
-    Parameters
-    ----------
-    task_extractor : Dict
-        used to match target tasks
-    origins: List[Metadata]
-        metadata is usually a class feature
-        label with a finite values.
-    """
-
-    def __init__(self, task_index, **kwargs):
-        print(task_index)
-        if isinstance(task_index, str):
-            task_index = FileOps.load(task_index)
-
-        self.base_model = kwargs.get("base_model")(num_class=31)
-
-        self.seen_task_key = KBResourceConstant.SEEN_TASK.value
-        self.task_group_key = KBResourceConstant.TASK_GROUPS.value
-        self.extractor_key = KBResourceConstant.EXTRACTOR.value
-
-        self.backup_model = kwargs.get('backup_model')
-        if not self.backup_model:
-            self.seen_extractor = task_index.get(self.seen_task_key).get(self.extractor_key)
-            if isinstance(self.seen_extractor, str):
-                self.seen_extractor = FileOps.load(self.seen_extractor)
-            self.seen_task_groups = task_index.get(self.seen_task_key).get(self.task_group_key)
-            self.seen_models = [task.model for task in self.seen_task_groups]
-        else:
-            self.backup_model = FileOps.download(self.backup_model)        
-
-        self.OOD_thresh = float(kwargs.get("OOD_thresh"))
-        self.ood_model = FileOps.load(kwargs.get("OOD_model"))
-        self.preprocess_func = kwargs.get("preprocess_func")
-
-    def __call__(self, samples: BaseDataSource, **
-                 kwargs) -> Tuple[BaseDataSource, BaseDataSource]:
-        '''
-        Parameters
-        ----------
-        samples : BaseDataSource
-            inference samples
-
-        Returns
-        -------
-        seen_task_samples: BaseDataSource
-        unseen_task_samples: BaseDataSource
-        '''
-        origin = kwargs.get("origin", "garden")
-        seen_task_samples = BaseDataSource(data_type=samples.data_type)
-        unseen_task_samples = BaseDataSource(data_type=samples.data_type)
-        seen_task_samples.x, unseen_task_samples.x = [], []
-
-        if not self.backup_model:
-            allocations = [self.seen_extractor.get(origin) for _ in samples.x]
-            samples, models = self._task_remodeling(samples=samples, mappings=allocations)
-        else:
-            models = [self.backup_model]
-            samples.inx = range(samples.num_examples())
-            samples = [samples]
-
-        tasks = []
-        for inx, df in enumerate(samples):
-            m = models[inx]
-            if not isinstance(m, Model):
-                continue
-            if isinstance(m.model, str):
-                evaluator = set_backend(estimator=self.base_model)
-                evaluator.load(m.model)
-            else:
-                evaluator = m.model
-            InD_list, OoD_list, pred = self.ood_predict(evaluator, df.x, **kwargs)
-            seen_task_samples.x.extend(InD_list)
-            unseen_task_samples.x.extend(OoD_list)
-            task = Task(entry=m.entry, samples=df)
-            task.result = pred
-            task.model = m
-            tasks.append(task)
-        res = self._inference_integrate(tasks)
-        return seen_task_samples, unseen_task_samples, res, tasks
-
-    def ood_predict(self, evaluator, samples, **kwargs):
-        data = self.preprocess_func(samples)
-        evaluator.estimator.validator.test_loader = DataLoader(
-            data,
-            batch_size=2,
-            shuffle=False,
-            pin_memory=True)
-
-        self.seg_model = evaluator.estimator.validator.model
-        self.data_loader = evaluator.estimator.validator.test_loader
-
-        OoD_list = []
-        InD_list = []
-
-        input = None
-        predictions = []
-        self.seg_model.eval()
-        for i, (sample, image_name) in enumerate(self.data_loader):
-            image = sample['image'].cuda()
-            with torch.no_grad():
-                output = self.seg_model(image)
-
-            torch.cuda.synchronize()
-
-            pred = output.data.cpu().numpy()
-            pred = np.argmax(pred, axis=1)
-
-            maxLogit = torch.max(output, 1)[0].unsqueeze(1)
-            maxLogit = self.batch_min_max(maxLogit)
-
-            softmaxDistance = self.get_softmaxDistance(output).unsqueeze(1)
-            maxLogit, softmaxDistance = maxLogit.mean(
-                1, keepdim=True), softmaxDistance.mean(
-                1, keepdim=True)
-            origin_shape = maxLogit.shape
-            maxLogit, softmaxDistance = maxLogit.flatten(), softmaxDistance.flatten()
-
-            effec_shape = maxLogit.shape[0]
-            if input == 'maxLogit':
-                temp_x = maxLogit.reshape(effec_shape, 1)
-            elif input == 'softmaxDistance':
-                temp_x = softmaxDistance.reshape(effec_shape, 1)
-            else:
-                temp_x = torch.cat([maxLogit.reshape(
-                    effec_shape, 1), softmaxDistance.reshape(effec_shape, 1)], dim=1)
-
-            OOD_pred = self.ood_model.predict(temp_x.cpu().numpy())
-            OOD_pred_show = OOD_pred + 1
-            OOD_pred_show = OOD_pred_show.reshape(origin_shape)
-            maxLogit = maxLogit.reshape(origin_shape)
-
-            for j in range(maxLogit.shape[0]):
-                OOD_score = (OOD_pred_show[j] == 1).sum(
-                ) / (OOD_pred_show[j] != 0).sum()
-                print('OOD_score:', OOD_score)
-                if OOD_score > self.OOD_thresh:
-                    OoD_list.append(samples[i])
-                else:
-                    InD_list.append(samples[i])
-                    predictions.append(pred)
-
-        return InD_list, OoD_list, predictions
-
-    def batch_min_max(self, img):
-        max_value = torch.amax(img, [1, 2, 3]).unsqueeze(dim=1)
-        min_value = torch.amin(img, [1, 2, 3]).unsqueeze(dim=1)
-
-        [b, n, h, w] = img.shape
-        img1 = img.reshape(b, -1)
-        img2 = (img1 - min_value) / (max_value - min_value)
-        img2 = img2.reshape([b, n, h, w])
-        return img2
-
-    def get_softmaxDistance(self, logits):
-        seg_softmax_out = torch.nn.Softmax(dim=1)(logits.detach())
-        distance, _ = torch.topk(seg_softmax_out, 2, dim=1)
-        max_softmaxLogit = distance[:, 0, :, :]
-        max2nd_softmaxLogit = distance[:, 1, :, :]
-        return max_softmaxLogit - max2nd_softmaxLogit
-
-    def _task_remodeling(self, samples: BaseDataSource, mappings: List):
-        """
-        Grouping based on assigned tasks
-        """
-        mappings = np.array(mappings)
-        data, models = [], []
-        d_type = samples.data_type
-        for m in np.unique(mappings):
-            task_df = BaseDataSource(data_type=d_type)
-            _inx = np.where(mappings == m)
-            if isinstance(samples.x, pd.DataFrame):
-                task_df.x = samples.x.iloc[_inx]
-            else:
-                task_df.x = np.array(samples.x)[_inx]
-            if d_type != "test":
-                if isinstance(samples.x, pd.DataFrame):
-                    task_df.y = samples.y.iloc[_inx]
-                else:
-                    task_df.y = np.array(samples.y)[_inx]
-            task_df.inx = _inx[0].tolist()
-            if samples.meta_attr is not None:
-                task_df.meta_attr = np.array(samples.meta_attr)[_inx]
-            data.append(task_df)
-            # TODO: if m is out of index
-            try:
-                model = self.seen_models[m]
-            except Exception as err:
-                print(f"self.models[{m}] not exists. {err}")
-                model = self.seen_models[0]
-            models.append(model)
-        return data, models
-
-    def _inference_integrate(self, tasks):
-        res = {}
-        for task in tasks:
-            res.update(dict(zip(task.samples.inx, task.result)))
-        return np.array([z[1]
-                        for z in sorted(res.items(), key=lambda x: x[0])])

lib/sedna/algorithms/unseen_task_detection/unseen_sample_recognition/unseen_sample_detection.py

@@ -20,7 +20,6 @@ class UnseenSampleDetection(threading.Thread):
     task_index: str or dict
         knowledge base index which includes indexes of tasks, samples and etc.
     '''
-    # MODEL_MANIPULATION_SEM = threading.Semaphore(1)
 
     def __init__(self, edge_knowledge_management, **kwargs):
         self.get_environ_varia()

lib/sedna/algorithms/unseen_task_detection/unseen_sample_recognition/unseen_sample_recognition.py

@@ -50,55 +50,10 @@ class SampleRegonitionDefault:
             else:
                 seen_task_samples.x = samples.x
                 unseen_task_samples.x = []
-            return seen_task_samples, unseen_task_samples
+            return seen_task_samples, unseen_task_samples, None, None
 
         sample_num = int(len(samples.x) / 2)
         seen_task_samples.x = samples.x[:sample_num]
         unseen_task_samples.x = samples.x[sample_num:]
 
-        return seen_task_samples, unseen_task_samples
-
-@ClassFactory.register(ClassType.UTD)
-class SampleRegonitionByRFNet:
-    '''
-    Divide inference samples into seen samples and unseen samples by confidence.
-
-    Parameters
-    ----------
-    task_index: str or dict
-        knowledge base index which includes indexes of tasks, samples and etc.
-    '''
-
-    def __init__(self, task_index, **kwargs):
-        if isinstance(task_index, str) and FileOps.exists(task_index):
-            self.task_index = FileOps.load(task_index)
-        else:
-            self.task_index = task_index
-
-        self.validator = kwargs.get("validator")
-
-    def __call__(self, samples: BaseDataSource, **
-                 kwargs) -> Tuple[BaseDataSource, BaseDataSource]:
-        '''
-        Parameters
-        ----------
-        samples : BaseDataSource
-            inference samples
-
-        Returns
-        -------
-        seen_task_samples: BaseDataSource
-        unseen_task_samples: BaseDataSource
-        '''
-        from torch.utils.data import DataLoader
-
-        self.validator.test_loader = DataLoader(
-            samples.x, batch_size=1, shuffle=False)
-
-        seen_task_samples = BaseDataSource(data_type=samples.data_type)
-        unseen_task_samples = BaseDataSource(data_type=samples.data_type)
-
-        seen_task_samples.x, unseen_task_samples.x = self.validator.task_divide()
-
-        return seen_task_samples, unseen_task_samples
-
+        return seen_task_samples, unseen_task_samples, None, None

lib/sedna/algorithms/unseen_task_detection/unseen_sample_recognition/unseen_sample_recognition_robotic.py

@@ -1,39 +0,0 @@
-from typing import Tuple
-
-from sedna.datasources import BaseDataSource
-from sedna.common.class_factory import ClassFactory, ClassType
-
-__all__ = ('SampleRegonitionRobotic')
-
-@ClassFactory.register(ClassType.UTD)
-class SampleRegonitionRobotic:
-    '''
-    Divide inference samples into seen samples and unseen samples
-
-    Parameters
-    ----------
-    task_index: str or dict
-        knowledge base index which includes indexes of tasks, samples and etc.
-    '''
-
-    def __init__(self, task_index, **kwargs):
-        pass
-
-    def __call__(self, samples: BaseDataSource) -> Tuple[BaseDataSource, BaseDataSource]:
-        '''
-        Parameters
-        ----------
-        samples : BaseDataSource
-            inference samples
-
-        Returns
-        -------
-        seen_task_samples: BaseDataSource
-        unseen_task_samples: BaseDataSource
-        '''
-        seen_task_samples = BaseDataSource(data_type=samples.data_type)
-        seen_task_samples.x = samples.x
-        unseen_task_samples = BaseDataSource(data_type=samples.data_type)
-        # unseen_task_samples.x = samples.x
-
-        return seen_task_samples, unseen_task_samples, None, None

lib/sedna/algorithms/knowledge_management/task_update_decision/task_update_decision.py → lib/sedna/core/lifelong_learning/knowledge_management/task_update_decision/task_update_decision.py

@@ -12,25 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""
-Divide multiple tasks based on data
-
-Parameters
-----------
-samples： Train data, see `sedna.datasources.BaseDataSource` for more detail.
-
-Returns
--------
-tasks: All tasks based on training data.
-task_extractor: Model with a method to predicting target tasks
-"""
-
-import time
-
 from sedna.common.file_ops import FileOps
 from sedna.datasources import BaseDataSource
 from sedna.common.class_factory import ClassType, ClassFactory
-from sedna.algorithms.seen_task_learning.artifact import Task
 
 __all__ = ('UpdateStrategyDefault', )
 
@@ -38,7 +22,8 @@ __all__ = ('UpdateStrategyDefault', )
 @ClassFactory.register(ClassType.KM)
 class UpdateStrategyDefault:
     """
-    Decide processing strategies for different tasks
+    Decide processing strategies for different tasks with labeled unseen samples.
+    Turn unseen samples to be seen.
 
     Parameters
     ----------

lib/sedna/core/lifelong_learning/lifelong_learning.py

@@ -19,15 +19,14 @@ from sedna.core.base import JobBase
 from sedna.common.file_ops import FileOps
 from sedna.common.constant import K8sResourceKind, K8sResourceKindStatus, KBResourceConstant
 from sedna.common.config import Context
-from sedna.datasources import BaseDataSource
 from sedna.common.class_factory import ClassType, ClassFactory
 from sedna.algorithms.seen_task_learning.seen_task_learning import SeenTaskLearning
 from sedna.algorithms.unseen_task_processing import UnseenTaskProcessing
 from sedna.algorithms.unseen_task_detection.unseen_sample_recognition.unseen_sample_detection import UnseenSampleDetection
 from sedna.service.client import KBClient
-from sedna.algorithms.knowledge_management.cloud_knowledge_management \
+from sedna.core.lifelong_learning.knowledge_management.cloud_knowledge_management \
     import CloudKnowledgeManagement
-from sedna.algorithms.knowledge_management.edge_knowledge_management \
+from sedna.core.lifelong_learning.knowledge_management.edge_knowledge_management \
     import EdgeKnowledgeManagement
 
 
@@ -289,13 +288,6 @@ class LifelongLearning(JobBase):
 
         seen_samples, unseen_samples = unseen_sample_re_recognition(train_data)
 
-        # TODO: retrain temporarily
-        # historical_data = self._fetch_historical_data(index_url)
-        # seen_samples.x = np.concatenate(
-        #     (historical_data.x, seen_samples.x, unseen_samples.x), axis=0)
-        # seen_samples.y = np.concatenate(
-        #     (historical_data.y, seen_samples.y, unseen_samples.y), axis=0)
-
         seen_samples.x = np.concatenate(
             (seen_samples.x, unseen_samples.x), axis=0)
         seen_samples.y = np.concatenate(
@@ -487,23 +479,4 @@ class LifelongLearning(JobBase):
         else:
             self.log.info(f"Deploy {index_file} to the edge.")
 
-        return res, index_file
-
-    def _fetch_historical_data(self, task_index):
-        if isinstance(task_index, str):
-            task_index = FileOps.load(task_index)
-
-        samples = BaseDataSource(data_type="train")
-
-        for task_group in task_index["seen_task"]["task_groups"]:
-            if isinstance(task_group.samples, BaseDataSource):
-                _samples = task_group.samples
-            else:
-                _samples = FileOps.load(task_group.samples.data_url)
-
-            samples.x = _samples.x if samples.x is None else np.concatenate(
-                (samples.x, _samples.x), axis=0)
-            samples.y = _samples.y if samples.y is None else np.concatenate(
-                (samples.y, _samples.y), axis=0)
-
-        return samples
+        return res, index_file