[1/3] Federation implementation

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]

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"
+```

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)

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

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()

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}")