New wrapper for anomaly detectors

src/mlpro_int_river/wrappers/changedetectors/__init__.py

@@ -0,0 +1 @@
+from .basics import *

src/mlpro_int_river/wrappers/changedetectors/basics.py

@@ -0,0 +1,199 @@
+## -------------------------------------------------------------------------------------------------
+## -- Project : MLPro - The integrative middleware framework for standardized machine learning
+## -- Package : mlpro_int_river.wrappers.changedetectors
+## -- Module  : basics.py
+## -------------------------------------------------------------------------------------------------
+## -- History :
+## -- yyyy-mm-dd  Ver.      Auth.    Description
+## -- 2025-08-12  0.0.0     DS       Creation
+## -------------------------------------------------------------------------------------------------
+
+"""
+Ver. 0.0.0 (2025-08-12)
+
+This module provides wrapper root classes from River to MLPro, specifically for anomaly detectors. 
+
+Learn more:
+https://www.riverml.xyz/
+
+"""
+
+import numpy as np
+from river.anomaly.base import AnomalyDetector as ADRiver
+from river.base import DriftDetector as DDRiver
+
+from mlpro.bf import Log, ParamError
+from mlpro.bf.streams import StreamTask, Instance
+from mlpro.oa.streams.tasks.changedetectors.anomalydetectors.instancebased import AnomalyDetectorIBPG
+from mlpro.oa.streams.tasks.changedetectors.driftdetectors.instancebased import DriftDetectorIB
+from mlpro.oa.streams.tasks.changedetectors.anomalydetectors.anomalies.instancebased import PointAnomaly
+from mlpro.oa.streams.tasks.changedetectors.driftdetectors.clusterbased.generic import DriftDetectorCBGenSingleGradient
+from mlpro_int_river.wrappers import WrapperRiver
+
+
+
+# Export list for public API
+__all__ = [ 'WrAnomalyDetectorRiver2MLPro',
+            'WrDriftDetectorRiver2MLPro' ]
+
+
+
+
+## -------------------------------------------------------------------------------------------------
+## -------------------------------------------------------------------------------------------------
+class WrAnomalyDetectorRiver2MLPro (AnomalyDetectorIBPG, WrapperRiver):
+
+    C_TYPE = 'Anomaly Detector (river)'
+
+## -------------------------------------------------------------------------------------------------
+    def __init__(self, 
+                 p_algo_river : ADRiver, 
+                 p_range_max=StreamTask.C_RANGE_THREAD, 
+                 p_ada = True, p_duplicate_data = False, 
+                 p_visualize = False, 
+                 p_logging=Log.C_LOG_ALL, 
+                 p_anomaly_buffer_size = 100, 
+                 p_thrs_inst = 0, 
+                 p_group_anomaly_det = True, 
+                 p_instance_buffer_size : int = 20,
+                 p_detection_steprate : int = 1,
+                 **p_kwargs):
+
+        WrapperRiver.__init__( self, p_logging = p_logging )
+
+        AnomalyDetectorIBPG.__init__( self, 
+                                      p_group_anomaly_det = p_group_anomaly_det, 
+                                      p_name = type(p_algo_river).__name__, 
+                                      p_range_max = p_range_max, 
+                                      p_ada = True, 
+                                      p_duplicate_data = p_duplicate_data, 
+                                      p_visualize = p_visualize, 
+                                      p_logging = p_logging, 
+                                      p_anomaly_buffer_size = p_anomaly_buffer_size,
+                                      p_thrs_inst = 1,
+                                      **p_kwargs )
+
+        if ( p_detection_steprate > p_instance_buffer_size ) or ( p_detection_steprate < 1 ):
+            raise ParamError('Please set the parameter "p_detection_steprate" >= 1 and <= "p_instance_buffer_size"')
+
+        self._algo_river                = p_algo_river
+        self._inst_buffer_size          = p_instance_buffer_size
+        self._detection_steprate        = p_detection_steprate
+        self._inst_counter              = 0
+
+        self._inst_data_buffer : np.ndarray  = None
+        self._inst_data_buffer_full : bool   = False
+        self._inst_ref_buffer : np.ndarray   = np.empty(self._inst_buffer_size, dtype = object)
+
+        self._inst_buffer_pos : int          = 0
+
+        self._block_mode = ( self._detection_steprate == self._inst_buffer_size )
+
+
+## -------------------------------------------------------------------------------------------------
+    def _detect(self, p_instance : Instance, **p_kwargs):
+
+
+        feature_data   = p_instance.get_feature_data()
+        feature_values = feature_data.get_values()
+        num_features   = feature_data.get_related_set().get_num_dim()
+        feature_names  = feature_data.get_dim_ids()
+
+        instance_dict = dict(zip(feature_names, feature_values))
+        score = self._algo_river.score_one(instance_dict)
+
+        if score > 0.8: 
+            anomaly = PointAnomaly( p_status=True,
+                                    p_tstamp=p_instance.tstamp,
+                                    p_visualize=self.get_visualization(),
+                                    p_raising_object=self,
+                                    p_instances=[p_instance])
+
+            self._raise_anomaly_event(p_anomaly=anomaly, p_instance=p_instance)
+
+        self._algo_river.learn_one(instance_dict)
+
+
+
+
+
+## -------------------------------------------------------------------------------------------------
+## -------------------------------------------------------------------------------------------------
+class WrDriftDetectorRiver2MLPro (DriftDetectorIB, WrapperRiver):
+
+    C_TYPE = 'Drift Detector (river)'
+
+## -------------------------------------------------------------------------------------------------
+    def __init__(self, 
+                 p_algo_river : DDRiver, 
+                 p_range_max=StreamTask.C_RANGE_THREAD, 
+                 p_ada = True, 
+                 p_duplicate_data = False, 
+                 p_visualize = False, 
+                 p_logging=Log.C_LOG_ALL, 
+                 p_drift_buffer_size = 100, 
+                 p_thrs_inst = 0, 
+                 p_instance_buffer_size : int = 20,
+                 p_detection_steprate : int = 1,
+                 p_feature_idx=0,
+                 p_feature_id=None,
+                 **p_kwargs):
+
+        WrapperRiver.__init__( self, p_logging = p_logging )
+
+        DriftDetectorIB.__init__( self, 
+                                  p_name = type(p_algo_river).__name__, 
+                                  p_range_max = p_range_max, 
+                                  p_ada = True, 
+                                  p_duplicate_data = p_duplicate_data, 
+                                  p_visualize = p_visualize, 
+                                  p_logging = p_logging, 
+                                  p_drift_buffer_size = p_drift_buffer_size,
+                                  p_thrs_inst = 1,
+                                  **p_kwargs )
+
+        if ( p_detection_steprate > p_instance_buffer_size ) or ( p_detection_steprate < 1 ):
+            raise ParamError('Please set the parameter "p_detection_steprate" >= 1 and <= "p_instance_buffer_size"')
+
+        self._algo_river                = p_algo_river
+        self._feature_idx               = p_feature_idx
+        self._feature_id                = p_feature_id
+        self._inst_buffer_size          = p_instance_buffer_size
+        self._detection_steprate        = p_detection_steprate
+        self._inst_counter              = 0
+
+        self._inst_data_buffer : np.ndarray  = None
+        self._inst_data_buffer_full : bool   = False
+        self._inst_ref_buffer : np.ndarray   = np.empty(self._inst_buffer_size, dtype = object)
+
+        self._inst_buffer_pos : int          = 0
+
+        self._block_mode = ( self._detection_steprate == self._inst_buffer_size )
+
+
+## -------------------------------------------------------------------------------------------------
+    def _detect(self, p_instance : Instance, **p_kwargs):
+
+        # Get input feature vector
+        feature_data = p_instance.get_feature_data()
+
+        # Determine feature value to monitor
+        if self._feature_id is not None:
+            value = feature_data.get_value(self._feature_id)
+        else:
+            value = feature_data.get_values()[self._feature_idx]
+
+        # Feed value into River drift detector
+        self._algo_river.update(value)
+
+        if self._algo_river.drift_detected:
+            drift = DriftDetectorCBGenSingleGradient(
+                p_status=True,
+                p_tstamp=p_instance.tstamp,
+                p_visualize=self.get_visualization(),
+                p_raising_object=self,
+                p_instances=[p_instance]
+            )
+
+            self._raise_drift_event(p_drift=drift, p_instance=p_instance)
+

test/howtos/oa/changedetection/anomalydetection/howto_oa_cd_ad_001_point_anomaly_detection_hst_1d.py

@@ -0,0 +1,158 @@
+## -------------------------------------------------------------------------------------------------
+## -- Project : MLPro - The integrative middleware framework for standardized machine learning
+## -- Package : mlpro_int_river
+## -- Module  : howto_oa_cd_ad_001_point_anomaly_detection_hst_1d.py
+## -------------------------------------------------------------------------------------------------
+## -- History :
+## -- yyyy-mm-dd  Ver.      Auth.    Description
+## -- 2025-08-12  0.0.0     DS       Creation
+## -------------------------------------------------------------------------------------------------
+
+"""
+Ver. 0.0.0 (2025-08-12)
+
+"""
+from river.anomaly import HalfSpaceTrees
+from mlpro_int_river.wrappers.changedetectors.basics import WrAnomalyDetectorRiver2MLPro
+
+from mlpro.bf import *
+from mlpro.bf.streams.streams import StreamMLProPOutliers
+from mlpro.oa.streams import OAStreamScenario, OAStreamWorkflow, OAStreamTask
+
+
+
+## ------------------------------------------------------------------------------------------------
+## ------------------------------------------------------------------------------------------------
+class ADPAScenarioHST1D (OAStreamScenario):
+
+    C_NAME = 'Point Anomaly Detection with Half Space Trees 1D'
+
+## ------------------------------------------------------------------------------------------------
+    def _setup(self, 
+               p_mode, 
+               p_ada : bool, 
+               p_visualize : bool, 
+               p_logging,
+               p_n_trees: int = 10,
+               p_height: int = 8,
+               p_window_size: int = 250,
+               p_seed: int = 20,
+               p_anomaly_buffer_size: int = 100,
+               p_instance_buffer_size: int = 50,
+               p_detection_steprate: int = 50, 
+               p_thrs_inst: int = 5,
+               p_thrs_clusters: int = 1,
+               **p_kwargs):
+
+        # 1 Get the native stream from MLPro stream provider
+        stream = StreamMLProPOutliers( p_functions = ['const'],#,'sin' , 'cos' , 'const'],
+                                       p_outlier_rate=0.02,
+                                       p_logging=p_logging,
+                                       p_seed= 20)
+
+        # 2 Creation of a workflow
+        workflow = OAStreamWorkflow( p_name='wf1',
+                                     p_range_max=OAStreamWorkflow.C_RANGE_NONE,
+                                     p_ada=p_ada,
+                                     p_visualize=p_visualize, 
+                                     p_logging=p_logging )
+
+        # 3 Instantiation of River 'HalfSpaceTrees' anomaly detector
+        river_hst = HalfSpaceTrees( n_trees = p_n_trees,
+                                    height = p_height,
+                                    window_size = p_window_size,
+                                    seed = p_seed )
+
+        # 4 Creation of tasks and add them to the workflow
+        anomalydetector = WrAnomalyDetectorRiver2MLPro( p_algo_river = river_hst,
+                                                        p_anomaly_buffer_size = p_anomaly_buffer_size,
+                                                        p_instance_buffer_size = p_instance_buffer_size,
+                                                        p_detection_steprate = p_detection_steprate,
+                                                        p_group_anomaly_det = False,
+                                                        p_visualize = p_visualize,
+                                                        p_logging = p_logging )
+
+        workflow.add_task(p_task = anomalydetector)
+
+        # 5 Return stream and workflow
+        return stream, workflow
+
+
+
+## -------------------------------------------------------------------------------------------------
+## -------------------------------------------------------------------------------------------------
+# 1 Preparation of demo/unit test mode
+if __name__ == "__main__":
+    # 1.1 Parameters for demo mode
+    cycle_limit             = 500
+    logging                 = Log.C_LOG_WE
+    step_rate               = 1
+    n_trees                 = 10
+    height                  = 8
+    window_size             = 250
+    seed                    = 20
+    anomaly_buffer_size     = 100
+    instance_buffer_size    = 50
+    detection_steprate      = 50
+
+    cycle_limit             = int(input(f'\nCycle limit (press ENTER for {cycle_limit}): ') or cycle_limit)
+    visualize               = input('Visualization Y/N (press ENTER for Y): ').upper() != 'N'
+    if visualize:
+        i = input(f'Visualization step rate (press ENTER for {step_rate}): ')
+        if i != '': step_rate = int(i)
+
+        i = input('Log level: "A"=All, "W"=Warnings only, "N"=Nothing (press ENTER for "W"): ').upper() 
+        if i == 'A': logging = Log.C_LOG_WE
+        elif i == 'N': logging = Log.C_LOG_NOTHING
+
+    n_trees                  = int(input(f'Algo HST: Number of trees to use (press ENTER for {n_trees}): ') or n_trees)
+    height                   = int(input(f'Algo HST: Height if each tree (press ENTER for {height}): ') or height)
+    window_size              = int(input(f'Algo HST: Number of observations (press ENTER for {window_size}): ') or window_size)
+    seed                     = int(input(f'Algo HST: Seed number (press ENTER for {seed}): ') or seed)
+
+
+else:
+    # 1.2 Parameters for internal unit test
+    cycle_limit             = 20
+    logging                 = Log.C_LOG_NOTHING
+    visualize               = False
+    step_rate               = 1
+    n_trees                 = 10
+    height                   = 8
+    window_size             = 250
+    seed                    = 20
+    anomaly_buffer_size     = 100
+    instance_buffer_size    = 10
+    detection_steprate      = 10
+
+
+# 2 Instantiate the stream scenario
+myscenario = ADPAScenarioHST1D( p_mode = Mode.C_MODE_REAL,
+                                p_cycle_limit = cycle_limit,
+                                p_visualize = visualize,
+                                p_logging = logging,
+                                p_n_trees = n_trees,
+                                p_height = height,
+                                p_window_size = window_size,  
+                                p_seed = seed,                           
+                                p_anomaly_buffer_size = anomaly_buffer_size,
+                                p_instance_buffer_size = instance_buffer_size,
+                                p_detection_steprate = detection_steprate )
+
+if visualize:
+    myscenario.init_plot( p_plot_settings=PlotSettings( p_view = PlotSettings.C_VIEW_ND,
+                                                        p_view_autoselect = True,
+                                                        p_step_rate = step_rate ) )
+
+
+# 3 Reset and run own stream scenario
+myscenario.reset()
+
+if __name__ == '__main__':
+    input('Press ENTER to start stream processing...')
+
+myscenario.run()
+
+
+if __name__ == '__main__':
+    input('Press ENTER to exit...')