diff --git a/agentMET4FOF/sensornetwork/base_classes.py b/agentMET4FOF/sensornetwork/base_classes.py new file mode 100644 index 00000000..8f89805b --- /dev/null +++ b/agentMET4FOF/sensornetwork/base_classes.py @@ -0,0 +1,70 @@ +from typing import Tuple, Union, List + +import numpy as np +import pandas as pd + +from agentMET4FOF.agents.metrological_base_agents import MetrologicalAgent +from agentMET4FOF.streams.metrological_base_streams import MetrologicalDataStreamMET4FOF + + +class SensorOnPlatform(MetrologicalDataStreamMET4FOF): + """Streaming data from a sensor located on a specified platform + + Parameters + ---------- + platform_name : str, optional + name of the platform on which the sensing unit is located + uncertainty : float + measurement uncertainty of the sensor. usually found on calibration certificate or tech specs + output_unit : str + SI unit of the sensor output + sensor_type : str, optional + type of sensor based on what is being measured + data_stream : Union[List, DataFrame, np.ndarray] + data stream of sensor measurements indexed by time, e.g. timestamps + """ + + def __init__( + self, uncertainty: float =0, platform_name=None, sensor_type=None, output_unit=None, data_stream: Union[List, pd.DataFrame, np.ndarray]=None + ): + self.uncertainty = uncertainty + self.output_unit = output_unit + self.platform = platform_name + self.sensor_type = sensor_type + super(SensorOnPlatform, self).__init__(value_unc=self.uncertainty, time_unc=0) + self.set_metadata( + self.platform+'_'+data_stream.columns.values[0], + "time", + "h", + self.sensor_type, + output_unit, + "Data Stream from Heat Meter Readings", + ) + self.set_data_source(quantities=data_stream, time=pd.DataFrame(data_stream.index.values)) + + +class SensorPlatform(MetrologicalAgent): + """A metrological agent representing a platform hosting one or more sensors in an IoT network + """ + + def init_parameters(self): + """Initialize the sensor agent + + Parameters + ----------- + uncertainty: np.float + The uncertainty of the sensor determined via a calibration + position: Union[Tuple[np.float, np.float], str] + The location of the sensor given either by explicit geographical coordinates or a string descriptor + """ + + super().init_parameters() + self.position = None + self.output_unit = None + self._stream = SensorOnPlatform(uncertainty=.01, platform_name="Heat Meter", sensor_type='Temperature', output_unit='°C', + data_stream=None) + + @property + def device_id(self): + return self._stream.metadata.metadata["device_id"] + diff --git a/agentMET4FOF/sensornetwork/rendezvous_agents.py b/agentMET4FOF/sensornetwork/rendezvous_agents.py new file mode 100644 index 00000000..48bea414 --- /dev/null +++ b/agentMET4FOF/sensornetwork/rendezvous_agents.py @@ -0,0 +1,262 @@ +from agentMET4FOF.agents import AgentMET4FOF +import math +from scipy.spatial import cKDTree +import numpy as np + + +class RendezvousAgent(AgentMET4FOF): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + + self.rendezvous_steps_list_buffer = None + self.spatial_threshold = None + self.time_threshold = None + + def init_parameters(self): + pass + + def get_time_distance(self, t1, t2): + return math.fabs(t2 - t1) + + def get_spatial_distance(self, coordinates_1, coordinates_2): + return math.sqrt( + (coordinates_2[0] - coordinates_1[0]) ** 2 + + (coordinates_2[1] - coordinates_1[1]) ** 2 + ) + + def find_neighbors( + self, + sensor_one_positions, + sensor_one_times, + sensor_two_positions, + sensor_two_times, + spatial_threshold, + time_threshold + ): + + positions_neighbor_pairs_set = self._find_isotropic_neighbors( + sensor_one_positions, sensor_two_positions, spatial_threshold + ) + times_neighbor_pairs_set = self._find_isotropic_neighbors( + sensor_one_times, sensor_two_times, time_threshold + ) + + neighbor_pairs_set = positions_neighbor_pairs_set.intersection(times_neighbor_pairs_set) + return neighbor_pairs_set + + def find_all_neighbors(self, sensor_positions_list, sensor_times_list, spatial_threshold, time_threshold): + assert len(sensor_positions_list) == len(sensor_times_list), (len(sensor_positions_list), len(sensor_times_list)) + n_sensors = len(sensor_positions_list) + + neighbor_pairs_set = set() + + for i in range(n_sensors): + for j in range(n_sensors): + if i <= j: + continue + + _neighbor_pairs_set_two_sensors = self.find_neighbors( + sensor_positions_list[i], + sensor_times_list[i], + sensor_positions_list[j], + sensor_times_list[j], + spatial_threshold, + time_threshold + ) + + _neighbor_pairs_set = { + (i, j, pairs[0], pairs[1]) for pairs in _neighbor_pairs_set_two_sensors + } + + neighbor_pairs_set.update(_neighbor_pairs_set) + + return neighbor_pairs_set + + def _find_isotropic_neighbors(self, data_one, data_two, r): + data_one = np.array(data_one) + data_two = np.array(data_two) + + if data_one.ndim == 1: + assert data_two.ndim == 1 + data_one = data_one.reshape(-1, 1) + data_two = data_two.reshape(-1, 1) + + data_one_tree = cKDTree(data_one) + data_two_tree = cKDTree(data_two) + + idx_to_neighbors = data_one_tree.query_ball_tree(data_two_tree, r) + pairs_list = set() + for i in range(len(idx_to_neighbors)): + for j in idx_to_neighbors[i]: + pairs_list.add((i, j)) + + return pairs_list + + # def get_number_neighbors(self, sensor_positions_list, sensor_times_list, spatial_threshold, time_threshold): + # assert len(sensor_positions_list) == len(sensor_times_list), (len(sensor_positions_list), len(sensor_times_list)) + # + # n_sensors = len(sensor_positions_list) + # + # cpt = 0 + # for i in range(n_sensors): + # for j in range(n_sensors): + # if i <= j: + # continue + # + # cpt += len(self.find_neighbors( + # sensor_positions_list[i], + # sensor_times_list[i], + # sensor_positions_list[j], + # sensor_times_list[j], + # spatial_threshold, + # time_threshold + # )) + # + # return cpt + + def on_received_message(self, message): + """ + User-defined method and is triggered to handle the message passed by Input. + + Parameters + ---------- + message : Dictionary + The message received is in form {'from':agent_name, 'data': data, + 'senderType': agent_class, 'channel':channel_name}. agent_name is the + name of the Input agent which sent the message data is the actual content + of the message. + + Assume that data is a dict: { + "sensor_position": list of size-2 tuples of floats, + "sensor_time": list of floats, + "other_sensor_position": list of size-2 tuples of floats, + "other_sensor_time": list of floats + } + + Find pairs of indices (i, j) such that the i-th measurement from the sensor is a rendezvous point with the j-th + measurement from the other sensor. + """ + + data = message["data"] + + sensor_position = data["sensor_position"] + sensor_time = data["sensor_time"] + + assert len(sensor_position) == len(sensor_time), "Incompatible sizes: {} vs {}".format( + len(sensor_position), len(sensor_time) + ) + + other_sensor_position = data["other_sensor_position"] + other_sensor_time = data["other_sensor_time"] + + assert len(other_sensor_position) == len(other_sensor_time), "Incompatible sizes: {} vs {}".format( + len(other_sensor_position), len(other_sensor_time) + ) + + self.rendezvous_steps_list_buffer = self.find_neighbors( + sensor_position, + sensor_time, + other_sensor_position, + other_sensor_time, + self.spatial_threshold, + self.time_threshold + ) + + def get_random_position_and_time_distance(self, sensor_positions_list, sensor_times_list): + n_sensors = len(sensor_positions_list) + + _random_idx_sensor = np.random.choice(n_sensors) + + _random_i, _random_j = np.random.choice(len(sensor_positions_list[_random_idx_sensor]), 2, replace=False) + + _random_position_i = sensor_positions_list[_random_idx_sensor][_random_i] + _random_position_j = sensor_positions_list[_random_idx_sensor][_random_j] + _random_position_distance = self.get_spatial_distance(_random_position_i, _random_position_j) + + _random_time_i = sensor_times_list[_random_idx_sensor][_random_i] + _random_time_j = sensor_times_list[_random_idx_sensor][_random_j] + _random_time_distance = self.get_time_distance(_random_time_i, _random_time_j) + + return _random_position_distance, _random_time_distance + + def select_threshold( + self, + sensor_positions_list, + sensor_times_list, + sensor_measurements_list, + n_rep_random=1000, + min_number_neighbors=20, + grid_size=10 + ): + """ + Select spatial and time thresholds for rendezvous points. + + Parameters + ---------- + sensor_positions_list : list of lists of size-2 tuples of floats + Coordinates (tuples of float) of measurements of all sensors. + sensor_times_list : list of lists of floats + Times (float) of measurements of all sensors. + sensor_measurements_list : list of lists of floats + Measurements from all sensors. + n_rep_random : int + Number of repetitions to compute medians which will serve for the center of the box of thresholds. + min_number_neighbors : int + Minimal number of neighbors to compute a correlation. TODO(): Use a Bayesian estimate. + grid_size : int + Correlation optimized on a grid of size grid_size x grid_size. + + The thresholds are chosen by optimizing the correlation between the measurements from rendezvous points. + TODO:() Use a Bayesian estimator instead of the parameter min_number_neighbors. + """ + + assert len(sensor_positions_list) == len(sensor_times_list), (len(sensor_positions_list), len(sensor_times_list)) + + distances_sample = [ + self.get_random_position_and_time_distance(sensor_positions_list, sensor_times_list) + for _ in range(n_rep_random) + ] + + # TODO:() Improve or expose quantiles + quantiles = np.quantile(distances_sample, [0.01, 0.5, 0.99], axis=0) + + spatial_threshold_bounds = [quantiles[0, 0], quantiles[2, 0]] + times_threshold_bounds = [quantiles[0, 1], quantiles[2, 1]] + + # TODO:() Use log-grids. + spatial_threshold_grid = np.linspace(spatial_threshold_bounds[0], spatial_threshold_bounds[1], grid_size) + time_threshold_grid = np.linspace(times_threshold_bounds[0], times_threshold_bounds[1], grid_size) + + correlations = np.zeros([grid_size, grid_size]) + + for i in range(grid_size): + for j in range(grid_size): + all_neighbors = self.find_all_neighbors(sensor_positions_list, sensor_times_list, spatial_threshold_grid[i], time_threshold_grid[j]) + if len(all_neighbors) <= min_number_neighbors: + # Improve convention? + correlations[i, j] = 0.0 + else: + data_i = [] + data_j = [] + + for neighbors in all_neighbors: + data_i.append( + sensor_measurements_list[neighbors[0]][neighbors[2]] + ) + + data_j.append( + sensor_measurements_list[neighbors[1]][neighbors[3]] + ) + + data_i = np.array(data_i) + data_j = np.array(data_j) + + # TODO:() Smooth the data? Use a Bayesian estimate? + correlations[i, j] = np.corrcoef(data_i.reshape(1, -1), data_j.reshape(1, -1))[0, 1] + + i_opt, j_opt = correlations.max(1).argmax(), correlations.max(0).argmax() + spatial_threshold_opt = spatial_threshold_grid[i_opt] + time_threshold_opt = time_threshold_grid[j_opt] + + self.spatial_threshold = spatial_threshold_opt + self.time_threshold = time_threshold_opt diff --git a/agentMET4FOF/streams/metrological_base_streams.py b/agentMET4FOF/streams/metrological_base_streams.py index 6a73f114..0aa1d98b 100644 --- a/agentMET4FOF/streams/metrological_base_streams.py +++ b/agentMET4FOF/streams/metrological_base_streams.py @@ -188,6 +188,49 @@ def _next_sample_generator(self, batch_size: int = 1) -> np.ndarray: return np.concatenate((_time, _time_unc, _value, _value_unc), axis=1) + def _next_sample_data_source( + self, batch_size: Optional[int] = 1 + ) -> np.ndarray: + """Internal method for fetching latest samples from a dataset. + Overrides :meth:`.DataStreamMET4FOF._next_sample_data_source`. Adds + time uncertainty ``ut`` and measurement uncertainty ``uv`` to sample + + """ + if batch_size < 0: + batch_size = self._quantities.shape[0] + + self._sample_idx += batch_size + + try: + self._current_sample_quantities = self._quantities[ + self._sample_idx - batch_size : self._sample_idx + ] + + # if target is available + if self._target is not None: + self._current_sample_target = self._target[ + self._sample_idx - batch_size : self._sample_idx + ] + else: + self._current_sample_target = None + + # if time is available + if self._time is not None: + self._current_sample_time = self._time[ + self._sample_idx - batch_size : self._sample_idx + ] + else: + self._current_sample_time = None + except IndexError: + self._current_sample_quantities = None + self._current_sample_target = None + self._current_sample_time = None + + _time_unc, _value_unc = (np.full_like(self._current_sample_time, fill_value=self.time_unc), + np.full_like(self._current_sample_quantities, fill_value=self.value_unc)) + + return np.concatenate((self._current_sample_time, _time_unc, self._current_sample_quantities, _value_unc), axis=1) + @property def value_unc(self) -> Union[float, Iterable[float]]: """Union[float, Iterable[float]]: uncertainties associated with the values""" diff --git a/agentMET4FOF_tutorials/tutorial_1_generator_agent.ipynb b/agentMET4FOF_tutorials/tutorial_1_generator_agent.ipynb index b1ff6f30..cbf29562 100644 --- a/agentMET4FOF_tutorials/tutorial_1_generator_agent.ipynb +++ b/agentMET4FOF_tutorials/tutorial_1_generator_agent.ipynb @@ -1,247 +1,72 @@ -{ - "cells": [ - { - "cell_type": "markdown", - "metadata": { - "pycharm": { - "name": "#%% md\n" - } - }, - "source": [ - "# Tutorial 1 - A simple pipeline to plot a signal\n", - "\n", - "First we define a simple pipeline of three agents. Two agents will generate a signal\n", - "(in our case based on the *SineGeneratorAgent*), one of which is a default signal and\n", - "one with customized parameters. The third one plots the signals on the dashboard,\n", - "thus needs to be of type *MonitorAgent* or *MetrologicalMonitorAgent*. Besides, the\n", - "outputs of the signal agents need to be bound to the *MonitorAgent*.\n", - "\n", - "\n", - "Each agent has an internal `current_state` which can be used as a switch to change the \n", - "behaviour of the agent. The available states are listed\n", - "[here](https://github.com/Met4FoF/agentMET4FOF/blob/a7e0007fcd460458d629546c99d68ef1f2079c11/agentMET4FOF/agents/base_agents.py#L160).\n", - "\n", - "As soon as all agents are initialized and the connections are set up, the agent \n", - "network is started by accordingly changing all agents' state simultaneously." - ] - }, - { - "cell_type": "code", - "execution_count": 2, - "metadata": { - "pycharm": { - "name": "#%%\n" - } - }, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "Error on connecting to existing name server at http://0.0.0.0:3333: Could not locate the name server!\n", - "Starting NameServer...\n", - "Broadcast server running on 0.0.0.0:9091\n", - "NS running on 0.0.0.0:3333 (0.0.0.0)\n", - "URI = PYRO:Pyro.NameServer@0.0.0.0:3333\n", - "INFO [2022-02-03 21:05:44.331772] (Default_Sine_generator): INITIALIZED\n", - "INFO [2022-02-03 21:05:44.360827] (Custom_Sine_generator): INITIALIZED\n", - "INFO [2022-02-03 21:05:44.386817] (Showcase_a_default_and_a_customized_sine_signal): INITIALIZED\n", - "[2022-02-03 21:05:44.399664] (Default_Sine_generator): Connected output module: Showcase_a_default_and_a_customized_sine_signal\n", - "[2022-02-03 21:05:44.405280] (Custom_Sine_generator): Connected output module: Showcase_a_default_and_a_customized_sine_signal\n", - "\n", - "|------------------------------------------------------------|\n", - "| |\n", - "| Your agent network is starting up. Open your browser and |\n", - "| visit the agentMET4FOF dashboard on http://127.0.0.1:8050/ |\n", - "| |\n", - "|------------------------------------------------------------|\n", - "\n", - "SET STATE: Running\n", - "[2022-02-03 21:05:45.338354] (Default_Sine_generator): Pack time: 0.001391\n", - "[2022-02-03 21:05:45.343559] (Default_Sine_generator): Sending: {'quantities': array([0.]), 'time': array([0.])}\n", - "[2022-02-03 21:05:45.347733] (Showcase_a_default_and_a_customized_sine_signal): Received: {'from': 'Default_Sine_generator', 'data': {'quantities': array([0.]), 'time': array([0.])}, 'senderType': 'SineGeneratorAgent', 'channel': 'default'}\n", - "[2022-02-03 21:05:45.350192] (Showcase_a_default_and_a_customized_sine_signal): Buffer: {'Default_Sine_generator': {'quantities': array([0.]), 'time': array([0.])}}\n", - "[2022-02-03 21:05:45.351553] (Showcase_a_default_and_a_customized_sine_signal): Tproc: 0.002991\n", - "[2022-02-03 21:05:45.375015] (Custom_Sine_generator): Pack time: 0.00176\n", - "[2022-02-03 21:05:45.378430] (Showcase_a_default_and_a_customized_sine_signal): Received: {'from': 'Custom_Sine_generator', 'data': {'quantities': array([0.]), 'time': array([0.])}, 'senderType': 'SineGeneratorAgent', 'channel': 'default'}\n", - "[2022-02-03 21:05:45.379148] (Custom_Sine_generator): Sending: {'quantities': array([0.]), 'time': array([0.])}\n", - "[2022-02-03 21:05:45.385440] (Showcase_a_default_and_a_customized_sine_signal): Buffer: {'Default_Sine_generator': {'quantities': array([0.]), 'time': array([0.])}, 'Custom_Sine_generator': {'quantities': array([0.]), 'time': array([0.])}}\n", - "[2022-02-03 21:05:45.386185] (Showcase_a_default_and_a_customized_sine_signal): Tproc: 0.007304\n", - "[2022-02-03 21:05:46.336481] (Default_Sine_generator): Pack time: 0.000488\n", - "[2022-02-03 21:05:46.340135] (Default_Sine_generator): Sending: {'quantities': array([0.38460898]), 'time': array([0.01])}\n", - "[2022-02-03 21:05:46.340536] (Showcase_a_default_and_a_customized_sine_signal): Received: {'from': 'Default_Sine_generator', 'data': {'quantities': array([0.38460898]), 'time': array([0.01])}, 'senderType': 'SineGeneratorAgent', 'channel': 'default'}\n", - "[2022-02-03 21:05:46.345115] (Showcase_a_default_and_a_customized_sine_signal): Buffer: {'Default_Sine_generator': {'quantities': array([0. , 0.38460898]), 'time': array([0. , 0.01])}, 'Custom_Sine_generator': {'quantities': array([0.]), 'time': array([0.])}}\n", - "[2022-02-03 21:05:46.345760] (Showcase_a_default_and_a_customized_sine_signal): Tproc: 0.004564\n", - "[2022-02-03 21:05:46.366489] (Custom_Sine_generator): Pack time: 0.000678\n", - "[2022-02-03 21:05:46.370212] (Custom_Sine_generator): Sending: {'quantities': array([0.28845673]), 'time': array([0.01])}\n", - "[2022-02-03 21:05:46.379294] (Showcase_a_default_and_a_customized_sine_signal): Received: {'from': 'Custom_Sine_generator', 'data': {'quantities': array([0.28845673]), 'time': array([0.01])}, 'senderType': 'SineGeneratorAgent', 'channel': 'default'}\n", - "[2022-02-03 21:05:46.403425] (Showcase_a_default_and_a_customized_sine_signal): Buffer: {'Default_Sine_generator': {'quantities': array([0. , 0.38460898]), 'time': array([0. , 0.01])}, 'Custom_Sine_generator': {'quantities': array([0. , 0.28845673]), 'time': array([0. , 0.01])}}\n", - "[2022-02-03 21:05:46.406775] (Showcase_a_default_and_a_customized_sine_signal): Tproc: 0.020121\n", - "[2022-02-03 21:05:47.336524] (Default_Sine_generator): Pack time: 0.000599\n", - "[2022-02-03 21:05:47.339019] (Default_Sine_generator): Sending: {'quantities': array([0.71004939]), 'time': array([0.02])}\n", - "[2022-02-03 21:05:47.339234] (Showcase_a_default_and_a_customized_sine_signal): Received: {'from': 'Default_Sine_generator', 'data': {'quantities': array([0.71004939]), 'time': array([0.02])}, 'senderType': 'SineGeneratorAgent', 'channel': 'default'}\n", - "[2022-02-03 21:05:47.344371] (Showcase_a_default_and_a_customized_sine_signal): Buffer: {'Default_Sine_generator': {'quantities': array([0. , 0.38460898, 0.71004939]), 'time': array([0. , 0.01, 0.02])}, 'Custom_Sine_generator': {'quantities': array([0. , 0.28845673]), 'time': array([0. , 0.01])}}\n", - "[2022-02-03 21:05:47.344878] (Showcase_a_default_and_a_customized_sine_signal): Tproc: 0.005222\n", - "[2022-02-03 21:05:47.365561] (Custom_Sine_generator): Pack time: 0.000555\n", - "[2022-02-03 21:05:47.368761] (Showcase_a_default_and_a_customized_sine_signal): Received: {'from': 'Custom_Sine_generator', 'data': {'quantities': array([0.53253704]), 'time': array([0.02])}, 'senderType': 'SineGeneratorAgent', 'channel': 'default'}\n", - "[2022-02-03 21:05:47.368108] (Custom_Sine_generator): Sending: {'quantities': array([0.53253704]), 'time': array([0.02])}\n", - "[2022-02-03 21:05:47.375238] (Showcase_a_default_and_a_customized_sine_signal): Buffer: {'Default_Sine_generator': {'quantities': array([0. , 0.38460898, 0.71004939]), 'time': array([0. , 0.01, 0.02])}, 'Custom_Sine_generator': {'quantities': array([0. , 0.28845673, 0.53253704]), 'time': array([0. , 0.01, 0.02])}}\n", - "[2022-02-03 21:05:47.375647] (Showcase_a_default_and_a_customized_sine_signal): Tproc: 0.005395\n", - "[2022-02-03 21:05:48.336290] (Default_Sine_generator): Pack time: 0.000678\n", - "[2022-02-03 21:05:48.339054] (Default_Sine_generator): Sending: {'quantities': array([0.92625524]), 'time': array([0.03])}\n", - "[2022-02-03 21:05:48.339381] (Showcase_a_default_and_a_customized_sine_signal): Received: {'from': 'Default_Sine_generator', 'data': {'quantities': array([0.92625524]), 'time': array([0.03])}, 'senderType': 'SineGeneratorAgent', 'channel': 'default'}\n", - "[2022-02-03 21:05:48.365860] (Custom_Sine_generator): Pack time: 0.000724\n", - "[2022-02-03 21:05:48.346087] (Showcase_a_default_and_a_customized_sine_signal): Buffer: {'Default_Sine_generator': {'quantities': array([0. , 0.38460898, 0.71004939, 0.92625524]), 'time': array([0. , 0.01, 0.02, 0.03])}, 'Custom_Sine_generator': {'quantities': array([0. , 0.28845673, 0.53253704]), 'time': array([0. , 0.01, 0.02])}}\n", - "[2022-02-03 21:05:48.369095] (Custom_Sine_generator): Sending: {'quantities': array([0.69469143]), 'time': array([0.03])}\n", - "[2022-02-03 21:05:48.346546] (Showcase_a_default_and_a_customized_sine_signal): Tproc: 0.006717\n", - "[2022-02-03 21:05:48.371092] (Showcase_a_default_and_a_customized_sine_signal): Received: {'from': 'Custom_Sine_generator', 'data': {'quantities': array([0.69469143]), 'time': array([0.03])}, 'senderType': 'SineGeneratorAgent', 'channel': 'default'}\n", - "[2022-02-03 21:05:48.388990] (Showcase_a_default_and_a_customized_sine_signal): Buffer: {'Default_Sine_generator': {'quantities': array([0. , 0.38460898, 0.71004939, 0.92625524]), 'time': array([0. , 0.01, 0.02, 0.03])}, 'Custom_Sine_generator': {'quantities': array([0. , 0.28845673, 0.53253704, 0.69469143]), 'time': array([0. , 0.01, 0.02, 0.03])}}\n", - "[2022-02-03 21:05:48.389764] (Showcase_a_default_and_a_customized_sine_signal): Tproc: 0.017948\n", - "[2022-02-03 21:05:49.334544] (Default_Sine_generator): Pack time: 0.000107\n", - "[2022-02-03 21:05:49.334972] (Default_Sine_generator): Sending: {'quantities': array([0.99996522]), 'time': array([0.04])}\n", - "[2022-02-03 21:05:49.335138] (Showcase_a_default_and_a_customized_sine_signal): Received: {'from': 'Default_Sine_generator', 'data': {'quantities': array([0.99996522]), 'time': array([0.04])}, 'senderType': 'SineGeneratorAgent', 'channel': 'default'}\n", - "[2022-02-03 21:05:49.335881] (Showcase_a_default_and_a_customized_sine_signal): Buffer: {'Default_Sine_generator': {'quantities': array([0. , 0.38460898, 0.71004939, 0.92625524, 0.99996522]), 'time': array([0. , 0.01, 0.02, 0.03, 0.04])}, 'Custom_Sine_generator': {'quantities': array([0. , 0.28845673, 0.53253704, 0.69469143]), 'time': array([0. , 0.01, 0.02, 0.03])}}\n", - "[2022-02-03 21:05:49.335959] (Showcase_a_default_and_a_customized_sine_signal): Tproc: 0.000742\n", - "[2022-02-03 21:05:49.363757] (Custom_Sine_generator): Pack time: 0.00011\n", - "[2022-02-03 21:05:49.364182] (Custom_Sine_generator): Sending: {'quantities': array([0.74997391]), 'time': array([0.04])}\n", - "[2022-02-03 21:05:49.364331] (Showcase_a_default_and_a_customized_sine_signal): Received: {'from': 'Custom_Sine_generator', 'data': {'quantities': array([0.74997391]), 'time': array([0.04])}, 'senderType': 'SineGeneratorAgent', 'channel': 'default'}\n", - "[2022-02-03 21:05:49.364982] (Showcase_a_default_and_a_customized_sine_signal): Buffer: {'Default_Sine_generator': {'quantities': array([0. , 0.38460898, 0.71004939, 0.92625524, 0.99996522]), 'time': array([0. , 0.01, 0.02, 0.03, 0.04])}, 'Custom_Sine_generator': {'quantities': array([0. , 0.28845673, 0.53253704, 0.69469143, 0.74997391]), 'time': array([0. , 0.01, 0.02, 0.03, 0.04])}}\n", - "[2022-02-03 21:05:49.365063] (Showcase_a_default_and_a_customized_sine_signal): Tproc: 0.000664\n", - "[2022-02-03 21:05:50.336098] (Default_Sine_generator): Pack time: 0.000511\n", - "[2022-02-03 21:05:50.338496] (Default_Sine_generator): Sending: {'quantities': array([0.91983974]), 'time': array([0.05])}\n", - "[2022-02-03 21:05:50.339476] (Showcase_a_default_and_a_customized_sine_signal): Received: {'from': 'Default_Sine_generator', 'data': {'quantities': array([0.91983974]), 'time': array([0.05])}, 'senderType': 'SineGeneratorAgent', 'channel': 'default'}\n", - "[2022-02-03 21:05:50.349178] (Showcase_a_default_and_a_customized_sine_signal): Buffer: {'Default_Sine_generator': {'quantities': array([0. , 0.38460898, 0.71004939, 0.92625524, 0.99996522,\n", - " 0.91983974]), 'time': array([0. , 0.01, 0.02, 0.03, 0.04, 0.05])}, 'Custom_Sine_generator': {'quantities': array([0. , 0.28845673, 0.53253704, 0.69469143, 0.74997391]), 'time': array([0. , 0.01, 0.02, 0.03, 0.04])}}\n", - "[2022-02-03 21:05:50.349932] (Showcase_a_default_and_a_customized_sine_signal): Tproc: 0.009823\n", - "[2022-02-03 21:05:50.365870] (Custom_Sine_generator): Pack time: 0.00076\n", - "[2022-02-03 21:05:50.379973] (Showcase_a_default_and_a_customized_sine_signal): Received: {'from': 'Custom_Sine_generator', 'data': {'quantities': array([0.68987981]), 'time': array([0.05])}, 'senderType': 'SineGeneratorAgent', 'channel': 'default'}\n", - "[2022-02-03 21:05:50.373635] (Custom_Sine_generator): Sending: {'quantities': array([0.68987981]), 'time': array([0.05])}\n", - "[2022-02-03 21:05:50.387528] (Showcase_a_default_and_a_customized_sine_signal): Buffer: {'Default_Sine_generator': {'quantities': array([0. , 0.38460898, 0.71004939, 0.92625524, 0.99996522,\n", - " 0.91983974]), 'time': array([0. , 0.01, 0.02, 0.03, 0.04, 0.05])}, 'Custom_Sine_generator': {'quantities': array([0. , 0.28845673, 0.53253704, 0.69469143, 0.74997391,\n", - " 0.68987981]), 'time': array([0. , 0.01, 0.02, 0.03, 0.04, 0.05])}}\n", - "[2022-02-03 21:05:50.388292] (Showcase_a_default_and_a_customized_sine_signal): Tproc: 0.007594\n", - "[2022-02-03 21:05:51.337162] (Default_Sine_generator): Pack time: 0.000738\n", - "[2022-02-03 21:05:51.342613] (Showcase_a_default_and_a_customized_sine_signal): Received: {'from': 'Default_Sine_generator', 'data': {'quantities': array([0.69820537]), 'time': array([0.06])}, 'senderType': 'SineGeneratorAgent', 'channel': 'default'}\n", - "[2022-02-03 21:05:51.342614] (Default_Sine_generator): Sending: {'quantities': array([0.69820537]), 'time': array([0.06])}\n", - "[2022-02-03 21:05:51.353736] (Showcase_a_default_and_a_customized_sine_signal): Buffer: {'Default_Sine_generator': {'quantities': array([0. , 0.38460898, 0.71004939, 0.92625524, 0.99996522,\n", - " 0.91983974, 0.69820537]), 'time': array([0. , 0.01, 0.02, 0.03, 0.04, 0.05, 0.06])}, 'Custom_Sine_generator': {'quantities': array([0. , 0.28845673, 0.53253704, 0.69469143, 0.74997391,\n", - " 0.68987981]), 'time': array([0. , 0.01, 0.02, 0.03, 0.04, 0.05])}}\n", - "[2022-02-03 21:05:51.365000] (Custom_Sine_generator): Pack time: 0.000373\n", - "[2022-02-03 21:05:51.366463] (Custom_Sine_generator): Sending: {'quantities': array([0.52365403]), 'time': array([0.06])}\n", - "[2022-02-03 21:05:51.354944] (Showcase_a_default_and_a_customized_sine_signal): Tproc: 0.007321\n", - "[2022-02-03 21:05:51.369610] (Showcase_a_default_and_a_customized_sine_signal): Received: {'from': 'Custom_Sine_generator', 'data': {'quantities': array([0.52365403]), 'time': array([0.06])}, 'senderType': 'SineGeneratorAgent', 'channel': 'default'}\n", - "[2022-02-03 21:05:51.374387] (Showcase_a_default_and_a_customized_sine_signal): Buffer: {'Default_Sine_generator': {'quantities': array([0. , 0.38460898, 0.71004939, 0.92625524, 0.99996522,\n", - " 0.91983974, 0.69820537]), 'time': array([0. , 0.01, 0.02, 0.03, 0.04, 0.05, 0.06])}, 'Custom_Sine_generator': {'quantities': array([0. , 0.28845673, 0.53253704, 0.69469143, 0.74997391,\n", - " 0.68987981, 0.52365403]), 'time': array([0. , 0.01, 0.02, 0.03, 0.04, 0.05, 0.06])}}\n", - "[2022-02-03 21:05:51.374676] (Showcase_a_default_and_a_customized_sine_signal): Tproc: 0.004771\n" - ] - } - ], - "source": [ - "# %load tutorial_1_generator_agent.py\n", - "from time import sleep\n", - "\n", - "import numpy as np\n", - "\n", - "from agentMET4FOF.network import AgentNetwork\n", - "from agentMET4FOF.agents import MonitorAgent, SineGeneratorAgent\n", - "\n", - "\n", - "def demonstrate_generator_agent_use() -> AgentNetwork:\n", - " # Start agent network server.\n", - " agent_network = AgentNetwork()\n", - "\n", - " # Initialize agents by adding them to the agent network.\n", - " default_sine_agent = agent_network.add_agent(\n", - " name=\"Default Sine generator\", agentType=SineGeneratorAgent\n", - " )\n", - "\n", - " # Custom parameters of the specified agentType can either be handed over as **kwargs\n", - " # in the network instance's add_agent() method, or...\n", - " custom_sine_agent = agent_network.add_agent(\n", - " name=\"Custom Sine generator\",\n", - " agentType=SineGeneratorAgent,\n", - " sfreq=75,\n", - " sine_freq=np.pi,\n", - " )\n", - " # ... in a separate call of the agent instance's init_parameters() method.\n", - " custom_sine_agent.init_parameters(\n", - " amplitude=0.75,\n", - " )\n", - "\n", - " monitor_agent = agent_network.add_agent(\n", - " name=\"Showcase a default and a customized sine signal\", agentType=MonitorAgent\n", - " )\n", - "\n", - " # Interconnect agents by either way:\n", - " # 1) by agent network.bind_agents(source, target).\n", - " agent_network.bind_agents(default_sine_agent, monitor_agent)\n", - "\n", - " # 2) by the agent.bind_output().\n", - " custom_sine_agent.bind_output(monitor_agent)\n", - "\n", - " # Set all agents' states to \"Running\".\n", - " agent_network.set_running_state()\n", - "\n", - " # Allow for shutting down the network after execution\n", - " return agent_network\n", - "\n", - "\n", - "if __name__ == \"__main__\":\n", - " signal_demo_network = demonstrate_generator_agent_use()\n", - " sleep(60)\n", - " signal_demo_network.shutdown()\n" - ] - } - ], - "metadata": { - "kernelspec": { - "display_name": "Python 3 (ipykernel)", - "language": "python", - "name": "python3" - }, - "language_info": { - "codemirror_mode": { - "name": "ipython", - "version": 3 - }, - "file_extension": ".py", - "mimetype": "text/x-python", - "name": "python", - "nbconvert_exporter": "python", - "pygments_lexer": "ipython3", - "version": "3.11.9" - }, - "varInspector": { - "cols": { - "lenName": 16, - "lenType": 16, - "lenVar": 40 - }, - "kernels_config": { - "python": { - "delete_cmd_postfix": "", - "delete_cmd_prefix": "del ", - "library": "var_list.py", - "varRefreshCmd": "print(var_dic_list())" - }, - "r": { - "delete_cmd_postfix": ") ", - "delete_cmd_prefix": "rm(", - "library": "var_list.r", - "varRefreshCmd": "cat(var_dic_list()) " - } - }, - "types_to_exclude": [ - "module", - "function", - "builtin_function_or_method", - "instance", - "_Feature" - ], - "window_display": false - } - }, - "nbformat": 4, - "nbformat_minor": 4 -} +#%% md +# Tutorial 1 - A simple pipeline to plot a signal + +First we define a simple pipeline of three agents. Two agents will generate a signal +(in our case based on the *SineGeneratorAgent*), one of which is a default signal and +one with customized parameters. The third one plots the signals on the dashboard, +thus needs to be of type *MonitorAgent* or *MetrologicalMonitorAgent*. Besides, the +outputs of the signal agents need to be bound to the *MonitorAgent*. + + +Each agent has an internal `current_state` which can be used as a switch to change the +behaviour of the agent. The available states are listed +[here](https://github.com/Met4FoF/agentMET4FOF/blob/a7e0007fcd460458d629546c99d68ef1f2079c11/agentMET4FOF/agents/base_agents.py#L160). + +As soon as all agents are initialized and the connections are set up, the agent +network is started by accordingly changing all agents' state simultaneously. +#%% +# %load tutorial_1_generator_agent.py +from time import sleep + +import numpy as np + +from agentMET4FOF.network import AgentNetwork +from agentMET4FOF.agents import MonitorAgent, SineGeneratorAgent + + +def demonstrate_generator_agent_use() -> AgentNetwork: + # Start agent network server. + agent_network = AgentNetwork() + + # Initialize agents by adding them to the agent network. + default_sine_agent = agent_network.add_agent( + name="Default Sine generator", agentType=SineGeneratorAgent + ) + + # Custom parameters of the specified agentType can either be handed over as **kwargs + # in the network instance's add_agent() method, or... + custom_sine_agent = agent_network.add_agent( + name="Custom Sine generator", + agentType=SineGeneratorAgent, + sfreq=75, + sine_freq=np.pi, + ) + # ... in a separate call of the agent instance's init_parameters() method. + custom_sine_agent.init_parameters( + amplitude=0.75, + ) + + monitor_agent = agent_network.add_agent( + name="Showcase a default and a customized sine signal", agentType=MonitorAgent + ) + + # Interconnect agents by either way: + # 1) by agent network.bind_agents(source, target). + agent_network.bind_agents(default_sine_agent, monitor_agent) + + # 2) by the agent.bind_output(). + custom_sine_agent.bind_output(monitor_agent) + + # Set all agents' states to "Running". + agent_network.set_running_state() + + # Allow for shutting down the network after execution + return agent_network + + +if __name__ == "__main__": + signal_demo_network = demonstrate_generator_agent_use() + sleep(60) + signal_demo_network.shutdown() + +#%% diff --git a/agentMET4FOF_tutorials/tutorial_4_metrological_streams.ipynb b/agentMET4FOF_tutorials/tutorial_4_metrological_streams.ipynb index ad0d704a..1ad6936d 100644 --- a/agentMET4FOF_tutorials/tutorial_4_metrological_streams.ipynb +++ b/agentMET4FOF_tutorials/tutorial_4_metrological_streams.ipynb @@ -1,188 +1,95 @@ -{ - "cells": [ - { - "cell_type": "markdown", - "metadata": { - "pycharm": { - "name": "#%% md\n" - } - }, - "source": [ - "# Tutorial 4 - A metrological datastream\n", - "\n", - "In this tutorial we introduce the new metrologically enabled agents. We initialize an\n", - "agent, which generates an infinite sine signal. The signal is generated from the\n", - "built-in class `MetrologicalSineGenerator` which delivers on each call one timestamp\n", - "and one value each with associated uncertainties.\n", - " \n", - "The _MetrologicalSineGeneratorAgent_ is based on the new class\n", - "_agentMET4FOF.metrological_agents.MetrologicalAgent_. We only adapt the\n", - "methods `init_parameters()` and `agent_loop()`. This we need to hand over an instance\n", - "of the signal generating class and to generate the actual samples. The rest of the\n", - "buffering and plotting logic is encapsulated inside the new base classes." - ] - }, - { - "cell_type": "code", - "execution_count": 2, - "metadata": { - "pycharm": { - "name": "#%%\n" - } - }, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "\n", - "|------------------------------------------------------------|\n", - "| |\n", - "| Your agent network is starting up. Open your browser and |\n", - "| visit the agentMET4FOF dashboard on http://127.0.0.1:8050/ |\n", - "| |\n", - "|------------------------------------------------------------|\n", - "\n", - "SET STATE: Running\n", - "[2022-02-03 21:09:14.499774] (SineGenerator): INITIALIZED\n", - "[2022-02-03 21:09:14.500004] (Metrological plot including measurement uncertainties): INITIALIZED\n", - "[2022-02-03 21:09:14.500050] (SineGenerator): Connected output module: Metrological plot including measurement uncertainties|\n" - ] - } - ], - "source": [ - "# %load tutorial_4_metrological_streams.py\n", - "from agentMET4FOF.network import AgentNetwork\n", - "from agentMET4FOF.metrological_agents import MetrologicalAgent, MetrologicalMonitorAgent\n", - "from agentMET4FOF.metrological_streams import (\n", - " MetrologicalDataStreamMET4FOF,\n", - " MetrologicalSineGenerator,\n", - ")\n", - "from agentMET4FOF.utils import Backend\n", - "\n", - "\n", - "class MetrologicalSineGeneratorAgent(MetrologicalAgent):\n", - " \"\"\"An agent streaming a sine signal\n", - "\n", - " Takes samples from an instance of :py:class:`MetrologicalSineGenerator` and pushes\n", - " them sample by sample to connected agents via its output channel.\n", - " \"\"\"\n", - "\n", - " # The datatype of the stream will be MetrologicalSineGenerator.\n", - " _stream: MetrologicalDataStreamMET4FOF\n", - "\n", - " def init_parameters(\n", - " self,\n", - " signal: MetrologicalDataStreamMET4FOF = MetrologicalSineGenerator(),\n", - " **kwargs\n", - " ):\n", - " \"\"\"Initialize the input data stream\n", - "\n", - " Parameters\n", - " ----------\n", - " signal : MetrologicalDataStreamMET4FOF\n", - " the underlying signal for the generator\n", - " \"\"\"\n", - " self._stream = signal\n", - " super().init_parameters()\n", - " self.set_output_data(channel=\"default\", metadata=self._stream.metadata)\n", - "\n", - " def agent_loop(self):\n", - " \"\"\"Model the agent's behaviour\n", - "\n", - " On state *Running* the agent will extract sample by sample the input\n", - " datastream's content and push it into its output buffer.\n", - " \"\"\"\n", - " if self.current_state == \"Running\":\n", - " self.set_output_data(channel=\"default\", data=self._stream.next_sample())\n", - " super().agent_loop()\n", - "\n", - "\n", - "def demonstrate_metrological_stream():\n", - "\n", - " # start agent network server\n", - " agent_network = AgentNetwork(backend=Backend.MESA)\n", - "\n", - " # Initialize signal generating class outside of agent framework.\n", - " signal = MetrologicalSineGenerator()\n", - "\n", - " # Initialize metrologically enabled agent taking name from signal source metadata.\n", - " source_name = signal.metadata.metadata[\"device_id\"]\n", - " source_agent = agent_network.add_agent(\n", - " name=source_name, agentType=MetrologicalSineGeneratorAgent\n", - " )\n", - " source_agent.init_parameters(signal)\n", - "\n", - " # Initialize metrologically enabled plotting agent.\n", - " monitor_agent = agent_network.add_agent(\n", - " \"Metrological plot including measurement uncertainties\",\n", - " agentType=MetrologicalMonitorAgent,\n", - " buffer_size=50,\n", - " )\n", - "\n", - " # Bind agents.\n", - " source_agent.bind_output(monitor_agent)\n", - "\n", - " # Set all agents states to \"Running\".\n", - " agent_network.set_running_state()\n", - "\n", - " # Allow for shutting down the network after execution.\n", - " return agent_network\n", - "\n", - "\n", - "if __name__ == \"__main__\":\n", - " demonstrate_metrological_stream()\n" - ] - } - ], - "metadata": { - "kernelspec": { - "display_name": "Python 3 (ipykernel)", - "language": "python", - "name": "python3" - }, - "language_info": { - "codemirror_mode": { - "name": "ipython", - "version": 3 - }, - "file_extension": ".py", - "mimetype": "text/x-python", - "name": "python", - "nbconvert_exporter": "python", - "pygments_lexer": "ipython3", - "version": "3.10.1" - }, - "varInspector": { - "cols": { - "lenName": 16, - "lenType": 16, - "lenVar": 40 - }, - "kernels_config": { - "python": { - "delete_cmd_postfix": "", - "delete_cmd_prefix": "del ", - "library": "var_list.py", - "varRefreshCmd": "print(var_dic_list())" - }, - "r": { - "delete_cmd_postfix": ") ", - "delete_cmd_prefix": "rm(", - "library": "var_list.r", - "varRefreshCmd": "cat(var_dic_list()) " - } - }, - "types_to_exclude": [ - "module", - "function", - "builtin_function_or_method", - "instance", - "_Feature" - ], - "window_display": false - } - }, - "nbformat": 4, - "nbformat_minor": 1 -} \ No newline at end of file +#%% md +# Tutorial 4 - A metrological datastream + +In this tutorial we introduce the new metrologically enabled agents. We initialize an +agent, which generates an infinite sine signal. The signal is generated from the +built-in class `MetrologicalSineGenerator` which delivers on each call one timestamp +and one value each with associated uncertainties. + +The _MetrologicalSineGeneratorAgent_ is based on the new class +_agentMET4FOF.metrological_agents.MetrologicalAgent_. We only adapt the +methods `init_parameters()` and `agent_loop()`. This we need to hand over an instance +of the signal generating class and to generate the actual samples. The rest of the +buffering and plotting logic is encapsulated inside the new base classes. +#%% +# %load tutorial_4_metrological_streams.py +from agentMET4FOF.network import AgentNetwork +from agentMET4FOF.metrological_agents import MetrologicalAgent, MetrologicalMonitorAgent +from agentMET4FOF.metrological_streams import ( + MetrologicalDataStreamMET4FOF, + MetrologicalSineGenerator, +) +from agentMET4FOF.utils import Backend + + +class MetrologicalSineGeneratorAgent(MetrologicalAgent): + """An agent streaming a sine signal + + Takes samples from an instance of :py:class:`MetrologicalSineGenerator` and pushes + them sample by sample to connected agents via its output channel. + """ + + # The datatype of the stream will be MetrologicalSineGenerator. + _stream: MetrologicalDataStreamMET4FOF + + def init_parameters( + self, + signal: MetrologicalDataStreamMET4FOF = MetrologicalSineGenerator(), + **kwargs + ): + """Initialize the input data stream + + Parameters + ---------- + signal : MetrologicalDataStreamMET4FOF + the underlying signal for the generator + """ + self._stream = signal + super().init_parameters() + self.set_output_data(channel="default", metadata=self._stream.metadata) + + def agent_loop(self): + """Model the agent's behaviour + + On state *Running* the agent will extract sample by sample the input + datastream's content and push it into its output buffer. + """ + if self.current_state == "Running": + self.set_output_data(channel="default", data=self._stream.next_sample()) + super().agent_loop() + + +def demonstrate_metrological_stream(): + + # start agent network server + agent_network = AgentNetwork(backend=Backend.MESA) + + # Initialize signal generating class outside of agent framework. + signal = MetrologicalSineGenerator() + + # Initialize metrologically enabled agent taking name from signal source metadata. + source_name = signal.metadata.metadata["device_id"] + source_agent = agent_network.add_agent( + name=source_name, agentType=MetrologicalSineGeneratorAgent + ) + source_agent.init_parameters(signal) + + # Initialize metrologically enabled plotting agent. + monitor_agent = agent_network.add_agent( + "Metrological plot including measurement uncertainties", + agentType=MetrologicalMonitorAgent, + buffer_size=50, + ) + + # Bind agents. + source_agent.bind_output(monitor_agent) + + # Set all agents states to "Running". + agent_network.set_running_state() + + # Allow for shutting down the network after execution. + return agent_network + + +if __name__ == "__main__": + demonstrate_metrological_stream()