diff --git a/src/detectmatelibrary/detectors/ecvc_detector.py b/src/detectmatelibrary/detectors/ecvc_detector.py new file mode 100644 index 0000000..30ca87b --- /dev/null +++ b/src/detectmatelibrary/detectors/ecvc_detector.py @@ -0,0 +1,131 @@ +from typing import Any, List + +from detectmatelibrary.common.detector import CoreDetector, CoreDetectorConfig +from detectmatelibrary.utils.data_buffer import BufferMode +from detectmatelibrary import schemas + +from scipy.stats import mode +from math import ceil +import numpy as np + + +class ECVCOp: + @staticmethod + def build_count_vec(input_: List[schemas.ParserSchema]) -> tuple[int, ...]: + sequence, n = [0], 0 + for in_ in input_: + event = in_["EventID"] + if n < event: + for _ in range(n, event): + sequence.append(0) + n = event + sequence[event] += 1 + + return tuple(sequence) + + @staticmethod + def build_one_vec(input_: List[schemas.ParserSchema], n: int) -> np.ndarray: + events = [in_["EventID"] for in_ in input_] + arr = np.zeros(n if n > (m := max(events) + 1) else m) + + for e in events: + arr[e] += 1 + + return arr + + @staticmethod + def init_count_matrix(seqs: set[tuple[int, ...]]) -> np.ndarray: + m, n = len(seqs), max([len(s) for s in seqs]) + matrix = np.zeros((m, n)) + + for i, seq in enumerate(seqs): + for j, c in enumerate(seq): + matrix[i, j] = c + + return matrix + + @staticmethod + def calculate_score(y: np.ndarray, matrix: np.ndarray) -> float: + pad = np.zeros((matrix.shape[0], y.shape[0] - matrix.shape[1])) + matrix_ = np.concat([matrix, pad], axis=1) + + score = np.inf + for m in matrix_: + dif = np.sum(np.abs(m - y)) + div = np.sum(np.max(np.concat([m[np.newaxis], y[np.newaxis]]).T, axis=1)) + score = score if score < (s := (dif / div)) else s + + return float(score) + + @staticmethod + def threshold_cal(y_s: np.ndarray, matrix: np.ndarray, method: str) -> float: + if method == "mean": + return float(np.mean([ECVCOp.calculate_score(y, matrix=matrix) for y in y_s])) + elif method == "mode": + return float(mode([ECVCOp.calculate_score(y, matrix=matrix) for y in y_s]).mode) + elif method == "default": + return 0.0 + + raise Exception("Method not supported") + + +class ECVCDetectorConfig(CoreDetectorConfig): + method_type: str = "ecvc_detector_detector" + window_size: int = 10 + validation_per: float = 0.2 + seed: int = 0 + threshold_method: str = "mean" + + +class ECVCDetector(CoreDetector): + def __init__( + self, + name: str = "ECVCDetector", + config: ECVCDetectorConfig | dict[str, Any] = ECVCDetectorConfig(), + ) -> None: + + if isinstance(config, dict): + config = ECVCDetectorConfig.from_dict(config, name) + self.config: ECVCDetectorConfig + + super().__init__( + name=name, + buffer_mode=BufferMode.WINDOW, + config=config, + buffer_size=config.window_size + ) + self.train_seqs: set[tuple[int, ...]] = set() + self.count_vecs: np.ndarray | None = None + self.threshold: float = 0 + + def train(self, input_: List[schemas.ParserSchema]) -> None: # type: ignore + self.train_seqs.add(ECVCOp.build_count_vec(input_)) + + def post_train(self) -> None: + train_idx = ceil(len(self.train_seqs) * (1 - self.config.validation_per)) + np.random.seed(self.config.seed) + matrix = ECVCOp.init_count_matrix(self.train_seqs)[np.random.permutation(len(self.train_seqs))] + + self.count_vecs, val = matrix[:train_idx], matrix[train_idx:] + if len(val) > 0: + self.threshold = ECVCOp.threshold_cal( + y_s=val, matrix=self.count_vecs, method=self.config.threshold_method + ) + self.train_seqs = set() + + def detect( + self, input_: List[schemas.ParserSchema], output_: schemas.DetectorSchema, # type: ignore + ) -> bool: + + if self.count_vecs is None: + return False + + score = ECVCOp.calculate_score(ECVCOp.build_one_vec( + input_, self.count_vecs.shape[1]), matrix=self.count_vecs + ) + if score > self.threshold: + output_["score"] = score + output_["description"] = "ECVC found an anominal sequence" + return True + + return False diff --git a/src/detectmatelibrary/detectors/scvs_detector.py b/src/detectmatelibrary/detectors/scvs_detector.py new file mode 100644 index 0000000..a8223de --- /dev/null +++ b/src/detectmatelibrary/detectors/scvs_detector.py @@ -0,0 +1,57 @@ +from typing import Any, List + +from detectmatelibrary.common.detector import CoreDetector, CoreDetectorConfig +from detectmatelibrary.utils.data_buffer import BufferMode +from detectmatelibrary import schemas + + +def build_count_vec(input_: List[schemas.ParserSchema]) -> tuple[int, ...]: + sequence, n = [0], 0 + for in_ in input_: + event = in_["EventID"] + if n < event: + for _ in range(n, event): + sequence.append(0) + n = event + sequence[event] += 1 + + return tuple(sequence) + + +class SCVSDetectorConfig(CoreDetectorConfig): + method_type: str = "scvs_detector" + window_size: int = 10 + + +class SCVSDetector(CoreDetector): + def __init__( + self, + name: str = "SCVSDetector", + config: SCVSDetectorConfig | dict[str, Any] = SCVSDetectorConfig(), + ) -> None: + + if isinstance(config, dict): + config = SCVSDetectorConfig.from_dict(config, name) + self.config: SCVSDetectorConfig + + super().__init__( + name=name, + buffer_mode=BufferMode.WINDOW, + config=config, + buffer_size=config.window_size + ) + self.train_seqs: set[tuple[int, ...]] = set() + + def train(self, input_: List[schemas.ParserSchema]) -> None: # type: ignore + self.train_seqs.add(build_count_vec(input_)) + + def detect( + self, input_: List[schemas.ParserSchema], output_: schemas.DetectorSchema, # type: ignore + ) -> bool: + + if build_count_vec(input_) not in self.train_seqs: + output_["score"] = 1. + output_["description"] = "Count vector not found" + return True + + return False diff --git a/tests/test_detectors/test_ecvc_detector.py b/tests/test_detectors/test_ecvc_detector.py new file mode 100644 index 0000000..7d3640e --- /dev/null +++ b/tests/test_detectors/test_ecvc_detector.py @@ -0,0 +1,103 @@ + +from detectmatelibrary.detectors.ecvc_detector import ECVCOp, ECVCDetectorConfig, ECVCDetector + +from detectmatelibrary import schemas + +import numpy as np + + +class TestECVCOP: + def test_build_count_vec(self): + input_ = [ + schemas.ParserSchema({"EventID": i}) for i in [0, 1, 4, 0] + ] + expected = tuple([2, 1, 0, 0, 1]) + + assert ECVCOp.build_count_vec(input_) == expected + + def test_build_one_vec(self): + input_ = [ + schemas.ParserSchema({"EventID": i}) for i in [0, 1, 4, 0] + ] + + expected = np.array([2, 1, 0, 0, 1, 0, 0, 0]) + assert np.array_equal(ECVCOp.build_one_vec(input_, 8), expected) + + expected = np.array([2, 1, 0, 0, 1]) + assert np.array_equal(ECVCOp.build_one_vec(input_, 3), expected) + + def test_init_count_matrix(self): + input_ = [ + (1, 0, 2, 1), (1, 2, 1), (0, 1, 4) + ] + expected = np.array([ + [1, 0, 2, 1], [1, 2, 1, 0], [0, 1, 4, 0] + ]) + + assert np.array_equal(ECVCOp.init_count_matrix(input_), expected) + + def test_calculate_score(self): + input_ = np.array([ + [1, 0, 2, 1], [1, 2, 1, 0], [0, 1, 4, 0] + ]) + y = np.array([0, 1, 4, 0, 0]) + + assert 0.0 == ECVCOp.calculate_score(y, matrix=input_) + + input_ = np.array([ + [1, 0, 2, 1], [1, 2, 1, 0], [0, 1, 4, 0] + ]) + y = np.array([0, 1, 4, 0]) + + assert 0.0 == ECVCOp.calculate_score(y, matrix=input_) + + def test_calculate_threshol(self): + matrix = np.array([ + [1, 0, 2, 1], [1, 2, 1, 0], [0, 1, 4, 0] + ]) + y = np.array([[1, 1, 1, 1], [0, 0, 0, 1]]) + + assert 0.0 == ECVCOp.threshold_cal(y, matrix=matrix, method="default") + assert 0.575 == ECVCOp.threshold_cal(y, matrix=matrix, method="mean") + assert 0.4 == ECVCOp.threshold_cal(y, matrix=matrix, method="mode") + + +class TestECVC: + def test_window_size(self): + ecvc = ECVCDetector(config=ECVCDetectorConfig(window_size=10)) + assert ecvc.get_window_size() == 10 + + ecvc = ECVCDetector(config=ECVCDetectorConfig(window_size=7)) + assert ecvc.get_window_size() == 7 + + def test_ecvc(self): + config = { + "detectors": { + "ECVCDetector": { + "method_type": "ecvc_detector_detector", + "window_size": 3, + "seed": 0, + "validation_per": 0., + "threshold_method": "mean", + "data_use_training": 30, + } + } + } + + ecvc = ECVCDetector(config=config) + + input_ = [] + for _ in range(10): + input_.extend([schemas.ParserSchema({"EventID": i}) for i in [0, 1, 4, 0]]) + + for in_ in input_: + ecvc.process(in_) + assert ecvc.get_state() == "Default" + + for in_ in input_[5:15]: + alert = ecvc.process(in_) + assert alert is None + + for in_ in [schemas.ParserSchema({"EventID": i}) for i in [4, 4, 4, 4, 4, 4, 1, 0, 1, 1]]: + alert = ecvc.process(in_) + assert alert is not None diff --git a/tests/test_detectors/test_scvs_detector.py b/tests/test_detectors/test_scvs_detector.py new file mode 100644 index 0000000..0a7e594 --- /dev/null +++ b/tests/test_detectors/test_scvs_detector.py @@ -0,0 +1,49 @@ + +from detectmatelibrary.detectors.scvs_detector import build_count_vec, SCVSDetector, SCVSDetectorConfig +from detectmatelibrary import schemas + + +class TestSCVSDetector: + def test_build_count_vec(self): + input_ = [ + schemas.ParserSchema({"EventID": i}) for i in [0, 1, 4, 0] + ] + expected = tuple([2, 1, 0, 0, 1]) + + assert build_count_vec(input_) == expected + + def test_window_size(self): + ecvc = SCVSDetector(config=SCVSDetectorConfig(window_size=10)) + assert ecvc.get_window_size() == 10 + + ecvc = SCVSDetector(config=SCVSDetectorConfig(window_size=7)) + assert ecvc.get_window_size() == 7 + + def test_scvs(self): + config = { + "detectors": { + "SCVSDetector": { + "method_type": "scvs_detector", + "window_size": 3, + "data_use_training": 30, + } + } + } + + scvs = SCVSDetector(config=config) + + input_ = [] + for _ in range(10): + input_.extend([schemas.ParserSchema({"EventID": i}) for i in [0, 1, 4, 0]]) + + for in_ in input_: + scvs.process(in_) + assert scvs.get_state() == "Default" + + for in_ in input_[5:15]: + alert = scvs.process(in_) + assert alert is None + + for in_ in [schemas.ParserSchema({"EventID": i}) for i in [4, 4, 4, 4, 4, 4, 1, 0, 1, 1]]: + alert = scvs.process(in_) + assert alert is not None