Merge pull request #101 from Met4FoF/sensor_issue_agents

Introduce sensor issue agents (Jitter and noise)

.gitignore

@@ -28,4 +28,5 @@ docs/CODE_OF_CONDUCT.md
 
 # Test clutter #
 ################
-test-reports/
+test-reports/
+.hypothesis/

agentMET4FOF/agents/base_agents.py

@@ -183,13 +183,9 @@ def init_buffer(self, buffer_size):
         return buffer
 
     def reset(self):
-        """
-        This method will be called on all agents when the global `reset_agents` is
-        called by the AgentNetwork and when the Reset button is clicked on the
-        dashboard.
+        """Reset the agent's states and parameters
 
-        Method to reset the agent's states and parameters. User can override this
-        method to reset the specific parameters.
+        User can override this method to reset the specific parameters.
         """
         self.log_info("RESET AGENT STATE")
         self.buffer.clear()

agentMET4FOF/agents/signal_agents.py

@@ -1,7 +1,12 @@
+from typing import Any, Dict, List, Optional, Union
+
+import numpy as np
+import pandas as pd
+
 from .base_agents import AgentMET4FOF
-from ..streams.signal_streams import SineGenerator
+from ..streams.signal_streams import SineGenerator, StaticSineWithJitterGenerator
 
-__all__ = ["SineGeneratorAgent"]
+__all__ = ["SineGeneratorAgent", "StaticSineWithJitterGeneratorAgent", "NoiseAgent"]
 
 
 class SineGeneratorAgent(AgentMET4FOF):
@@ -11,6 +16,8 @@ class SineGeneratorAgent(AgentMET4FOF):
     to connected agents via its output channel.
     """
 
+    _sine_stream: SineGenerator
+
     def init_parameters(self, sfreq=500, sine_freq=5, amplitude=1, initial_phase=0):
         """Initialize the input data
 
@@ -43,4 +50,108 @@ def agent_loop(self):
         """
         if self.current_state == "Running":
             sine_data = self._sine_stream.next_sample()  # dictionary
-            self.send_output(sine_data["quantities"])
+            self.send_output(sine_data)
+
+
+class StaticSineWithJitterGeneratorAgent(AgentMET4FOF):
+    """An agent streaming a pre generated sine signal of fixed length with jitter
+
+    Takes samples from the :py:mod:`StaticSineGeneratorWithJitter` and pushes them
+    sample by sample to connected agents via its output channel.
+    """
+
+    _sine_stream: StaticSineWithJitterGenerator
+
+    def init_parameters(
+        self, num_cycles: Optional[int] = 1000, jitter_std: Optional[float] = 0.02
+    ):
+        r"""Initialize the pre generated sine signal of fixed length with jitter
+
+        Initialize the static input data as an instance of the
+        :class:`StaticSineWithJitterGenerator` class with the provided parameters.
+
+        Parameters
+        ----------
+        num_cycles : int, optional
+            numbers of cycles, determines the signal length by :math:`\pi \cdot
+            num_cycles`, defaults to 1000
+        jitter_std : float, optional
+            the standard deviation of the distribution to randomly draw jitter from,
+            defaults to 0.02
+        """
+        self._sine_stream = StaticSineWithJitterGenerator(
+            num_cycles=num_cycles, jitter_std=jitter_std
+        )
+
+    def agent_loop(self):
+        """Extract sample by sample the input data stream's content and push it"""
+        if self.current_state == "Running":
+            sine_data = self._sine_stream.next_sample()  # dictionary
+            self.send_output(sine_data)
+
+
+class NoiseAgent(AgentMET4FOF):
+    """An agent adding white noise to the incoming signal"""
+    _noise_std: float
+
+    @property
+    def noise_std(self):
+        """Standard deviation of the distribution to randomly draw noise from"""
+        return self._noise_std
+
+    def init_parameters(self, noise_std: Optional[float] = 0.05):
+        """Initialize the noise's standard deviation
+
+        Parameters
+        ----------
+        noise_std : float, optional
+            the standard deviation of the distribution to randomly draw noise from,
+            defaults to 0.05
+        """
+        self._noise_std = noise_std
+
+    def on_received_message(self, message: Dict[str, Any]):
+        """Add noise to the received message's data
+
+        Parameters
+        ----------
+        message : Dictionary
+            the received message in the expected form::
+
+            dict like {
+                "from": "<valid agent name>"
+                "data": <time series data as a list, np.ndarray or pd.Dataframe> or
+                    dict like {
+                        "quantities": <time series data as a list, np.ndarray or
+                            pd.Dataframe>,
+                        "target": <target labels as a list, np.ndarray or pd.Dataframe>,
+                        "time": <time stamps as a list, np.ndarray or pd.Dataframe of
+                            float or np.datetime64>
+                    }
+                "senderType": <any subclass of AgentMet4FoF>,
+                "channel": "<channel name>"
+                }
+        """
+
+        def _compute_noisy_signal_from_clean_signal(
+            clean_signal: Union[List, np.ndarray, pd.DataFrame]
+        ):
+            return np.random.normal(
+                loc=clean_signal,
+                scale=self._noise_std,
+            )
+
+        if self.current_state == "Running":
+            data_in_message = message["data"].copy()
+            if isinstance(data_in_message, (list, np.ndarray, pd.DataFrame)):
+                self.send_output(
+                    _compute_noisy_signal_from_clean_signal(data_in_message)
+                )
+            if isinstance(data_in_message, dict):
+                fully_assembled_resulting_data = message["data"].copy()
+                fully_assembled_resulting_data[
+                    "quantities"
+                ] = _compute_noisy_signal_from_clean_signal(
+                    data_in_message["quantities"]
+                )
+                self.send_output(fully_assembled_resulting_data)