diff --git a/docs/diagrams.drawio b/docs/diagrams.drawio index 61c4bf68..3444a1ca 100644 --- a/docs/diagrams.drawio +++ b/docs/diagrams.drawio @@ -1,108 +1,154 @@ - - + + - - + + - - + + - - + + - - + + - - + + - + + + + + + + - - - - - - + + - + - - - - - - + + - - + + + + + - - + + - - + + - + + + + - - + + - + - - + + - + - - + + - - + + + + + - - + + + + + + + + + + - + + + + + + + + + + + + + + + + + + + + + + + + + + + + - - + + - + - - - - - - + + - + - - - - - - + + + + + + + + + + + + + + - + @@ -215,25 +261,25 @@ - + - + - + - + - + @@ -242,8 +288,110 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + - + @@ -280,16 +428,16 @@ - + - + - + diff --git a/docs/examples/others/federation.py b/docs/examples/others/federation.py new file mode 100644 index 00000000..8e21ffa2 --- /dev/null +++ b/docs/examples/others/federation.py @@ -0,0 +1,88 @@ + +# --8<-- [start:example_1] +from detectmatelibrary.common.core import CoreComponent +import struct + + +class NewComponent(CoreComponent): # Inherent from CoreComponent + def __init__(self, elems): + self.elems = elems + super().__init__(name="FedExample") + + def aggregate_strategy(self, components): + final_list = [] + for component in components: + final_list.extend(component.elems) + + final_list = list(set(final_list)) + for component in components: + component.elems = final_list + + def to_binary(self): + return struct.pack(f">{len(self.elems)}h", *self.elems) + + def from_binary(self, binary): + num_ints = len(binary) // 2 + elems = list(struct.unpack(f">{num_ints}h", binary)) + return NewComponent(elems=elems) + + +# --8<-- [end:example_1] +# --8<-- [start:example_2] +detector1 = NewComponent([1, 2, 3]) +detector2 = NewComponent([4, 5]) +detector3 = NewComponent([6]) + +detector1 + detector2 + detector3 + +detector2.aggregate() # Detector 2 is used as centralize node + +print("Dectector 3", detector3.elems) # All detectors have been updated + +# --8<-- [end:example_2] + +# --8<-- [start:example_3] +detector1 = NewComponent([1, 2, 3]) +detector2 = NewComponent([4, 5]) +detector3 = NewComponent([6]) + +detector1 + detector2 + detector3 + +detector2.aggregate() # Detector 2 is used as centralize node + +print("Dectector 3", detector3.elems) # All detectors have been updated +# --8<-- [end:example_3] + +# --8<-- [start:example_4] +detector1 = NewComponent([1, 2, 3]) +detector2 = NewComponent([4, 5]) +detector3 = NewComponent([6]) + +detector1.stack([detector2, detector3]) + +detector2.aggregate() +print("Detector 3", detector3.elems) # It is not longer combine but stack, so it will not work + +detector1.aggregate() +print("Detector 3", detector3.elems) # Now it will work + +# --8<-- [end:example_4] + +# --8<-- [start:example_5] +detector1 = NewComponent([1, 2, 3]) + +binary2 = NewComponent([4, 5]).to_binary() +binary3 = (detector3 := NewComponent([6])).to_binary() +print("Binary of Detector 3", binary3) + +detector1.stack([binary2, binary3]) +output = detector1.aggregate(unstack=True) # unstack = True will free memory + +print("Detector 1", detector1.elems) # Detector 1 has been updated +print("Output binary", output) # Output that we send to other componets + +# We update detector 3 now +print("Detector 3", detector3.elems) # Now detector 3 is not in the share memory so it will not work +detector3 = detector3.from_binary(output) +print("Detector 3", detector3.elems) # Now it will work +# --8<-- [end:example_5] diff --git a/docs/federation.md b/docs/federation.md new file mode 100644 index 00000000..d531b5a4 --- /dev/null +++ b/docs/federation.md @@ -0,0 +1,63 @@ +# Federation + +This section explains how to use the federation setup. For a component to support federation, it must implement the following methods: + +```python +def to_binary(self) -> bytes | None: + """(Federation only) Serialize to bytes for federation operations.""" + +def from_binary(self, binary: bytes) -> object: + """(Federation only) Deserialize from bytes for federation operations.""" + +def aggregate_strategy(self, components: set["FedOperations"]) -> None: + """(Federation only) Define how to aggregate a set of federated components.""" +``` + +There are two main ways to use federation: + +- **Combine first**: This can only be used when all components run locally. The main idea is to simplify the process by allowing components to share memory. +- **Stack later**: A more standard federated approach where the "weights" or state of each component are combined at the end. + +## Example class + +For all the examples below, we will use this code: + +```python +--8<-- "docs/examples/others/federation.py:example_1" +``` + +## Combine first + +The diagram below shows the workflow: + +![combine](img/fed_combine_first.png) + +Example 1: + +```python +--8<-- "docs/examples/others/federation.py:example_2" +``` + +Example 2: + +```python +--8<-- "docs/examples/others/federation.py:example_3" +``` + +## Stack + +The diagram below shows the workflow: + +![stack](img/fed_stack_later.png) + +Example 1: + +```python +--8<-- "docs/examples/others/federation.py:example_4" +``` + +Example 2: + +```python +--8<-- "docs/examples/others/federation.py:example_5" +``` diff --git a/docs/img/fed_combine_first.png b/docs/img/fed_combine_first.png new file mode 100644 index 00000000..4a923e09 Binary files /dev/null and b/docs/img/fed_combine_first.png differ diff --git a/docs/img/fed_stack_later.png b/docs/img/fed_stack_later.png new file mode 100644 index 00000000..42a2edf1 Binary files /dev/null and b/docs/img/fed_stack_later.png differ diff --git a/docs/overall_architecture.md b/docs/overall_architecture.md index 49f8e821..77850f6e 100644 --- a/docs/overall_architecture.md +++ b/docs/overall_architecture.md @@ -70,14 +70,42 @@ class Component(CoreComponent): * Default: the component is just processing data """ + def export_state( + self, path: str | None = None, storage_options: dict[str, Any] | None = None, + ) -> bytes | None: + """Export the current state if persistency class was implemented""" + + def import_state( + self, path: str | bytes, storage_options: dict[str, Any] | None = None + ) -> None: + """Import the current state if persistency class was implemented""" + def process(self, data: BaseSchema | bytes) -> BaseSchema | bytes | None: """Process the data in a stream fashion (Defined in the CoreComponent)""" def get_config(self) -> Dict[str, Any]: - """"Get the configuration of the component (Defined in the CoreComponent)""" + """Get the configuration of the component (Defined in the CoreComponent)""" def update_config(self, new_config: Dict[str, Any]) -> None: - """"Update the configuration of the component (Defined in the CoreComponent)""" + """Update the configuration of the component (Defined in the CoreComponent)""" + + def get_window_size(self) -> int: + """Get window size of the data buffer""" + + def stack(self, other: object | list[object | bytes] | bytes) -> None: + """(Federation only) stack multiple components for federation tasks""" + + def aggregate(self, unstack: bool = False) -> None | bytes: + """(Federation only) aggregate multiple components""" + + def to_binary(self) -> bytes | None: + """(Federation only) fill it to be compatible with federation ops""" + + def from_binary(self, binary: bytes) -> object: + """(Federation only) fill it to be compatible with federation ops""" + + def aggregate_strategy(self, components: set["FedOperations"]) -> None: + """(Federation only) fill it to be compatible with federation ops""" ``` Go back [Index](index.md) diff --git a/mkdocs.yml b/mkdocs.yml index 1e901d3c..1faf60e8 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -14,6 +14,7 @@ nav: - Basic concepts: basic_idea.md - Overall architecture: overall_architecture.md - Schemas: schemas.md + - Federation: federation.md - Parsers: parsers.md - Detectors: detectors.md - Alert Aggregation: alert_aggregator.md diff --git a/src/detectmatelibrary/common/_core_op/_basic_component.py b/src/detectmatelibrary/common/_core_op/_basic_component.py new file mode 100644 index 00000000..143a8e99 --- /dev/null +++ b/src/detectmatelibrary/common/_core_op/_basic_component.py @@ -0,0 +1,55 @@ +from detectmatelibrary.utils.persistency.component_interfaces import Stoppable +from detectmatelibrary.schemas import BaseSchema + +from detectmatelibrary.common._config import BasicConfig + +from typing import Any, Dict, List + + +class Component: + """Empty methods.""" + def __init__( + self, + name: str, + type_: str = "Core", + config: BasicConfig = BasicConfig(), + ) -> None: + self.name, self.type_, self.config = name, type_, config + self.saver: Stoppable | None = None + + def __repr__(self) -> str: + return f"<{self.type_}> {self.name}: {self.config}" + + def run( + self, input_: List[BaseSchema] | BaseSchema, output_: BaseSchema + ) -> bool: + return False + + def train( + self, input_: List[BaseSchema] | BaseSchema, + ) -> None: + pass + + def configure( + self, input_: List[BaseSchema] | BaseSchema, + ) -> None: + pass + + def set_configuration(self) -> None: + pass + + def post_train(self) -> None: + pass + + def get_config(self) -> Dict[str, Any]: + return self.config.get_config() + + def update_config(self, new_config: Dict[str, Any]) -> None: + self.config.update_config(new_config) + + def __enter__(self) -> "Component": + return self + + def __exit__(self, *_: Any) -> None: + if self.saver is not None: + self.saver.stop() diff --git a/src/detectmatelibrary/common/_core_op/_fed_component.py b/src/detectmatelibrary/common/_core_op/_fed_component.py new file mode 100644 index 00000000..eef4900d --- /dev/null +++ b/src/detectmatelibrary/common/_core_op/_fed_component.py @@ -0,0 +1,108 @@ + +import warnings +from typing import Self, overload + + +class IncompatibleFed(Exception): + def __init__(self) -> None: + super().__init__("Instances are incompatible") + + +class _CompOp: + @staticmethod + def is_compatible(main_inst: object, other_inst: object) -> None: + if not isinstance(other_inst, type(main_inst)): + raise IncompatibleFed() + + @staticmethod + def reset(main_inst: object, attr: str) -> None: + main_inst.__setattr__(attr, {main_inst}) + + @staticmethod + def combine(main_inst: object, attr: str, other_inst: object) -> None: + _CompOp.is_compatible(main_inst, other_inst) + set_main: set[object] = getattr(main_inst, attr) + + set_main.update(getattr(other_inst, attr)) + for elem in set_main: + getattr(elem, attr).update(set_main) + + @staticmethod + def uncombine(main_inst: object, attr: str, other_inst: object) -> None: + _CompOp.is_compatible(main_inst, other_inst) + set_main: set[object] = getattr(main_inst, attr) + + for elem in list(set_main): + if elem == other_inst: + _CompOp.reset(other_inst, attr) + else: + elem._components.remove(other_inst) # type: ignore + + @staticmethod + def stack(main_inst: object, attr: str, list_other_inst: list[object]) -> None: + set_main: set[object] = getattr(main_inst, attr) + for other_inst in list_other_inst: + _CompOp.is_compatible(main_inst, other_inst) + set_main.add(other_inst) + + +class FedOperations: + """Operations related to the federation learning / agregation.""" + __COMPONENT: str = "_components" + + def __init__(self) -> None: + self._components: set["FedOperations"] + _CompOp.reset(self, self.__COMPONENT) + + def __add__(self, other: object) -> Self: + """Add other components to do the aggregation in a combine first + approach.""" + _CompOp.combine(self, attr=self.__COMPONENT, other_inst=other) + + return self + + def __sub__(self, other: object) -> Self: + """Remove other components to do the aggregation in a combine first + approach.""" + _CompOp.uncombine(self, attr=self.__COMPONENT, other_inst=other) + + return self + + @overload + def stack(self, other: bytes | list[bytes]) -> None: + pass + + @overload + def stack(self, other: object | list[object]) -> None: + """Stack other components to do the aggregation in a stack later + approach.""" + pass + + def stack(self, other: object | list[object | bytes] | bytes) -> None: + if not isinstance(other, list): + other = [other] + _CompOp.stack( + self, attr=self.__COMPONENT, list_other_inst=[ + self.from_binary(inst) if isinstance(inst, bytes) else inst for inst in other + ] + ) + + def aggregate(self, unstack: bool = False) -> None | bytes: + self.aggregate_strategy(self._components) + + if unstack: + _CompOp.reset(self, self.__COMPONENT) + + return self.to_binary() + + def to_binary(self) -> bytes | None: + warnings.warn("To binary not implemented, return None") + return None + + def from_binary(self, binary: bytes) -> object: + warnings.warn(f"From binary not implemented, return None for {binary!r}") + return None + + def aggregate_strategy(self, components: set["FedOperations"]) -> None: + """Aggregation strategy use by the component.""" + warnings.warn(f"No strategy found, aggregations does nothing for {components}") diff --git a/src/detectmatelibrary/common/core.py b/src/detectmatelibrary/common/core.py index 7d52d54c..75a3c645 100644 --- a/src/detectmatelibrary/common/core.py +++ b/src/detectmatelibrary/common/core.py @@ -1,5 +1,7 @@ from detectmatelibrary.common._core_op._fit_logic import FitLogicState, StatesL from detectmatelibrary.common._core_op._schema_pipeline import SchemaPipeline +from detectmatelibrary.common._core_op._fed_component import FedOperations +from detectmatelibrary.common._core_op._basic_component import Component from detectmatelibrary.common._core_op._fit_logic import FitLogic from detectmatelibrary.utils.data_buffer import DataBuffer, ArgsBuffer, BufferMode @@ -12,10 +14,9 @@ from detectmatelibrary.tools.logging import logger, setup_logging -from typing import Any, Dict, List +from typing import Any from detectmatelibrary.utils.persistency.component_interfaces import PersistencyOp -from detectmatelibrary.utils.persistency.component_interfaces import Stoppable setup_logging() @@ -43,7 +44,7 @@ def __iter__(self) -> "TrainBuffer": return self -# Core component skeleton structure ################################################ +# Core component ################################################ class CoreConfig(BasicConfig): start_id: int = 10 @@ -52,58 +53,7 @@ class CoreConfig(BasicConfig): use_config_data_as_training: bool = True -class Component: - """Empty methods.""" - def __init__( - self, - name: str, - type_: str = "Core", - config: CoreConfig = CoreConfig(), - ) -> None: - self.name, self.type_, self.config = name, type_, config - self.saver: Stoppable | None = None - - def __repr__(self) -> str: - return f"<{self.type_}> {self.name}: {self.config}" - - def run( - self, input_: List[BaseSchema] | BaseSchema, output_: BaseSchema - ) -> bool: - return False - - def train( - self, input_: List[BaseSchema] | BaseSchema, - ) -> None: - pass - - def configure( - self, input_: List[BaseSchema] | BaseSchema, - ) -> None: - pass - - def set_configuration(self) -> None: - pass - - def post_train(self) -> None: - pass - - def get_config(self) -> Dict[str, Any]: - return self.config.get_config() - - def update_config(self, new_config: Dict[str, Any]) -> None: - self.config.update_config(new_config) - - def __enter__(self) -> "Component": - return self - - def __exit__(self, *_: Any) -> None: - if self.saver is not None: - self.saver.stop() - - -# Core component ################################################ - -class CoreComponent(Component): +class CoreComponent(Component, FedOperations): """Base class for all components in the system.""" def __init__( self, @@ -114,7 +64,9 @@ def __init__( input_schema: type[BaseSchema] = BaseSchema, output_schema: type[BaseSchema] = BaseSchema ) -> None: - super().__init__(name=name, type_=type_, config=config) + Component.__init__(self, name=name, type_=type_, config=config) + FedOperations.__init__(self) + self.config: CoreConfig self.input_schema, self.output_schema = input_schema, output_schema self.data_buffer = DataBuffer(args_buffer) diff --git a/src/detectmatelibrary/common/deeplearning_detector.py b/src/detectmatelibrary/common/deeplearning_detector.py index 1189483e..25a88b58 100644 --- a/src/detectmatelibrary/common/deeplearning_detector.py +++ b/src/detectmatelibrary/common/deeplearning_detector.py @@ -7,6 +7,7 @@ from detectmatelibrary import schemas from typing import Any +import logging class DeepLearningDetectorConfig(CoreDetectorConfig): @@ -72,8 +73,8 @@ def post_train(self) -> None: if "top_k" in self.stats: self.top_k = int(self.stats["top_k"]) - print(self.model) - print("Top k assigned", self.top_k) + logging.info(self.model) + logging.info(f"Top k assigned {self.top_k}") def detect( self, diff --git a/tests/test_common/test_core_federation.py b/tests/test_common/test_core_federation.py new file mode 100644 index 00000000..826fe406 --- /dev/null +++ b/tests/test_common/test_core_federation.py @@ -0,0 +1,156 @@ +from detectmatelibrary.common._core_op._fed_component import IncompatibleFed +from detectmatelibrary.common.core import CoreComponent + +import struct + +import pytest + + +class DummyComponent(CoreComponent): + pass + + +class DummyComponent2(CoreComponent): + pass + + +class TestJoinOp: + def test_add(self) -> None: + component1 = CoreComponent(name="comp_1") + component2 = CoreComponent(name="comp_2") + component3 = CoreComponent(name="comp_3") + + component2 + component3 + assert len(component2._components) == 2 + assert component3._components == component2._components + + component1 = component1 + component3 + assert len(component1._components) == 3 + assert component1._components == component2._components + assert component1._components == component3._components + + def test_sub(self) -> None: + component1 = CoreComponent(name="comp_1") + component2 = CoreComponent(name="comp_2") + component3 = CoreComponent(name="comp_3") + + (component1 + component2 + component3) - component3 + assert len(component2._components) == 2 + assert component1._components == component2._components + assert component3._components == {component3} + + def test_incompatible(self) -> None: + component1 = DummyComponent2(name="comp_1") + component2 = DummyComponent(name="comp_2") + + with pytest.raises(IncompatibleFed): + component1 + component2 + with pytest.raises(IncompatibleFed): + component1 - component2 + + def test_stack(self) -> None: + component1 = CoreComponent(name="comp_1") + component2 = CoreComponent(name="comp_2") + component3 = CoreComponent(name="comp_3") + component4 = CoreComponent(name="comp_4") + + component1.stack(component2) + assert len(component1._components) == 2 + assert component1._components != component2._components + + component1.stack([component3, component4]) + assert len(component1._components) == 4 + + +class DummyAppendList(CoreComponent): + def __init__( + self, elems: list[str], name: str = "test", *args, **kwargs + ) -> None: + super().__init__(name, *args, **kwargs) + self.elems = elems + + def aggregate_strategy(self, components): + final_list = [] + for component in components: + final_list.extend(component.elems) + + final_list = list(set(final_list)) + for component in components: + component.elems = final_list + + def to_binary(self): + return struct.pack(f">{len(self.elems)}h", *self.elems) + + def from_binary(self, binary): + num_ints = len(binary) // 2 + elems = list(struct.unpack(f">{num_ints}h", binary)) + return DummyAppendList(elems=elems) + + +class DummyAppendListEmpty(CoreComponent): + def __init__( + self, elems: list[str], name: str = "test", *args, **kwargs + ) -> None: + super().__init__(name, *args, **kwargs) + self.elems = elems + + +class TestFedComponent: + def test_basic_aggregation(self) -> None: + comp1 = DummyAppendList(elems=[1, 2]) + comp2 = DummyAppendList(elems=[3, 4]) + + comp1.aggregate() + assert set(comp1.elems) == {1, 2} + + (comp1 + comp2).aggregate() + assert set(comp1.elems) == {1, 2, 3, 4} + + def test_stack_basic_aggregation(self) -> None: + comp1 = DummyAppendList(elems=[1, 2]) + comp2 = DummyAppendList(elems=[3, 4]) + comp3 = DummyAppendList(elems=[5]) + + comp1.stack([comp2, comp3]) + comp1.aggregate(unstack=True) + assert set(comp1.elems) == {1, 2, 3, 4, 5} + assert len(comp1._components) == 1 + + comp1.stack([comp2, comp3]) + comp1.aggregate(unstack=False) + assert set(comp1.elems) == {1, 2, 3, 4, 5} + assert len(comp1._components) == 3 + + def test_sanity_check(self) -> None: + comp1 = DummyAppendList(elems=[1, 2]) + comp2 = comp1.from_binary(comp1.to_binary()) + + assert comp2.elems == [1, 2] + + def test_stack_binary_aggregation(self) -> None: + comp1 = DummyAppendList(elems=[1, 2]) + comp2 = DummyAppendList(elems=[3, 4]) + comp3 = DummyAppendList(elems=[5]) + + comp1.stack([comp2.to_binary(), comp3.to_binary()]) + comp1.aggregate(unstack=True) + assert set(comp1.elems) == {1, 2, 3, 4, 5} + + comp1.stack([comp2.to_binary(), comp3.to_binary()]) + output = comp1.aggregate(unstack=True) + assert set(comp1.elems) == {1, 2, 3, 4, 5} + assert isinstance(output, bytes) + + def test_empty_feed_fields(self) -> None: + comp1 = DummyAppendListEmpty(elems=[1, 2]) + comp2 = DummyAppendListEmpty(elems=[3, 4]) + comp3 = DummyAppendListEmpty(elems=[5]) + + with pytest.warns(UserWarning): + comp1.to_binary() + + with pytest.warns(UserWarning): + comp1.from_binary(b"") + + with pytest.warns(UserWarning): + comp1.aggregate_strategy({comp1, comp2, comp3})