agentMET4FOF agents

Base agents

class agentMET4FOF.agents.base_agents.AgentMET4FOF(name='', host=None, serializer=None, transport=None, attributes=None, backend=Backend.OSBRAIN, mesa_model=<class 'mesa.model.Model'>)

Base class for all agents with specific functions to be overridden/supplied by user.

Behavioral functions for users to provide are init_parameters, agent_loop and on_received_message. Communicative functions are bind_output, unbind_output and send_output. To learn more about the idea behind multi-agent systems (MAS) see [Bang2019].

_bind_output(output_agent, channel='default')

Internal method which implements the logic for connecting this agent, to the output_agent.

_convert_matplotlib_fig(fig: Figure, mode: str = 'image')

Convert matplotlib figure to be rendered by the dashboard

_convert_to_plotly(matplotlib_fig: Figure)

Internal method to convert matplotlib figure to plotly figure

Parameters:

matplotlib_fig (plt.Figure) – Matplotlib figure to be converted

_fig_to_uri(matplotlib_fig: Figure)

Internal method to convert matplotlib figure to base64 uri image for display

Parameters:

matplotlib_fig (plt.Figure) – Matplotlib figure to be converted

_get_metadata(data)

Internal helper function for getting the data type & dimensions of data. This is for update_output_channels_info()

static _is_type_message(data: Any) → bool

Internal method to check if the data carries signature of an agent message

Parameters:

data – Data to be checked for type

Returns:

result

Return type:

boolean

_remove_methods(cls)

Remove methods from the other backends base class from the current agent

_update_output_channels_info(data, channel)

Internal method to update the dict of output_channels_info. This is used in conjunction with send_output().

Checks and records data type & dimension and channel name If the data is nested within dict, then it will search deeper and subsequently record the info of each inner hierarchy

Parameters:

• data – data to be checked for type & dimension

• channel (str) – name of channel to be recorded

agent_loop()

User defined method for the agent to execute for loop_wait seconds specified either in self.loop_wait or explicitly via init_agent_loop( loop_wait)

To start a new loop, call init_agent_loop(loop_wait) on the agent. Example of usage is to check the current_state of the agent and send data periodically.

bind_output(output_agent, channel='default')

Forms Output connection with another agent

Any call on send_output will reach this newly binded agent. Adds the agent to its list of Outputs.

Parameters:

• output_agent (AgentMET4FOF or list) – Agent(s) to be binded to this agent’s output channel

• channel (str or list of str) – Specific name of the channel(s) to be subscribed to. (Default = “data”)

buffer_clear(agent_name: str | None = None)

Empties buffer which is a dict indexed by the agent_name.

Parameters:

agent_name (str, optional) – Key of the memory dict, which can be the name of input agent, or self.name. If not supplied (default), we assume to clear the entire memory.

buffer_filled(agent_name=None)

Checks whether the internal buffer has been filled to the maximum allowed specified by self.buffer_size

Parameters:

agent_name (str) – Index of the buffer which is the name of input agent.

Returns:

status of buffer filled

Return type:

boolean

buffer_store(agent_from: str, data=None, concat_axis=0)

Updates data stored in self.buffer with the received message

Checks if sender agent has sent any message before If it did,then append, otherwise create new entry for it

Parameters:

• agent_from (str) – Name of agent sender

• data – Any supported data which can be stored in dict as buffer. See AgentBuffer for details for more information.

• concat_axis (int, optional) – axis to concatenate on with the buffering for numpy arrays. Default is 0.

get_attr(attr)

Return the specified attribute of the agent.

Parameters:

name – Name of the attribute to be retrieved.

handle_process_data(message)

Internal method to handle incoming message before calling on_received_message

If current_state is either Stop or Reset, it will terminate early before entering on_received_message.

init_agent(buffer_size=1000, log_mode=True)

Internal initialization to set up the agent

This mainly sets the dictionaries of inputs, outputs and the public address.

Inputs

Dictionary of Agents connected to its input channels. Messages will arrive from agents in this dictionary. Automatically updated when bind_output() function is called

Type:

dict

Outputs

Dictionary of Agents connected to its output channels. Messages will be sent to agents in this dictionary. Automatically updated when bind_output() function is called

Type:

dict

PubAddr_alias

Name of Publish address socket

Type:

str

PubAddr

Publish address socket handle

Type:

str

AgentType

Name of class

Type:

str

current_state

Current state of agent. Can be used to define different states of operation such as “Running”, “Idle, “Stop”, etc.. Users will need to define their own flow of handling each type of self.current_state in the agent_loop

Type:

str

loop_wait

The interval to wait between loop. Call init_agent_loop to restart the timer or set the value of loop_wait in init_parameters when necessary.

Type:

int

buffer_size

The total number of elements to be stored in the agent buffer When total elements exceeds this number, the latest elements will be replaced with the incoming data elements

Type:

int

init_agent_loop(loop_wait: float | None = None)

Initiates the agent loop, which iterates every loop_wait seconds

Stops every timers and initiate a new loop.

Parameters:

loop_wait (float, optional) – The wait between each iteration of the loop, defaults to 1.0

init_buffer(buffer_size)

A method to initialise the buffer. By overriding this method, user can provide a custom buffer, instead of the regular AgentBuffer. This can be used, for example, to provide a MetrologicalAgentBuffer in the metrological agents.

init_parameters(**kwargs)

User provided function to initialize parameters of choice.

log_info(message)

Prints logs to be saved into logfile with Logger Agent

Parameters:

message (str) – Message to be logged to the internal Logger Agent

on_connect_output(output_agent)

This method is called whenever an agent is connected to its output

This can be for example, to send metadata or ping to the output agent.

on_received_message(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.

pack_data(data, channel='default')

Internal method to pack the data content into a dictionary before sending out.

Special case : if the data is already a message, then the from and senderType will be altered to this agent, without altering the data and channel within the message this is used for more succinct data processing and passing.

Parameters:

• data (argument) – Data content to be packed before sending out to agents.

• channel (str) – Key of dictionary which stores data

Returns:

• Packed message data (dict of the form {‘from’:agent_name, ‘data’: data,)

• ’senderType’ (agent_class, ‘channel’:channel_name}.)

reset()

Reset the agent’s states and parameters

User can override this method to reset the specific parameters.

respond_reply_attr_(message_data)

Response to a reply of setting attribute

respond_request_attr_(attribute: str)

Response to a request of attribute from input agents.

This agent reply with the requested attribute if it has it.

respond_request_method_(message_data: dict)

Response to a request of executing method from input agents.

This agent will execute the method with the provided parameters of the method.

send_output(data, channel='default')

Sends message data to all connected agents in self.Outputs.

Output connection can first be formed by calling bind_output. By default calls pack_data(data) before sending out. Can specify specific channel as opposed to ‘default’ channel.

Parameters:

• data (argument) – Data content to be sent out

• channel (str) – Key of message dictionary which stores data

Returns:

message – {‘from’:agent_name, ‘data’: data, ‘senderType’: agent_class, ‘channel’:channel_name}.

Return type:

dict

send_plot(fig: Figure | Dict[str, Figure], mode: str = 'image')

Sends plot to agents connected to this agent’s Output channel.

This method is different from send_output which will be sent to through the ‘plot’ channel to be handled.

Tradeoffs between “image” and “plotly” modes are that “image” are more stable and “plotly” are interactive. Note not all (complicated) matplotlib figures can be converted into a plotly figure.

Parameters:

• fig (matplotlib.figure.Figure or dict of matplotlib.figure.Figure) – Alternatively, multiple figures can be nested in a dict (with any preferred keys) e.g {“Temperature”:matplotlib.Figure, “Acceleration”:matplotlib.Figure}

• mode (str) – “image” - converts into image via encoding at base64 string. “plotly” - converts into plotly figure using mpl_to_plotly Default: “image”

Returns:

graph

Return type:

str or plotly figure or dict of one of those converted figure(s)

send_request_attribute(attribute: str)

Send a request of attribute to output agents.

Output agents will reply with the requested attribute if they have.

send_request_method(method: str, **method_params)

Send a request of executing methods to output agents.

Output agents will respond by calling the method.

send_set_attr(attr: str, value)

Sends a message to set the attr of another agent to that of value.

Parameters:

• attr (str) – The variable name of the output agent to be set.

• value – The value of the variable to be set

set_attr(**kwargs)

Set object attributes.

Parameters:

kwargs ([name, value]) – Keyword arguments will be used to set the object attributes.

shutdown()

Cleanly stop and shut down the agent assuming the agent is running.

Will let the main thread do the tear down.

step()

Used for MESA backend only. Behaviour on every update step.

stop_agent_loop()

Stops agent_loop from running

Note that the agent will still be responding to messages.

unbind_output(output_agent)

Remove existing output connection with another agent

This reverses the bind_output method.

Parameters:

output_agent (AgentMET4FOF) – Agent binded to this agent’s output channel

class agentMET4FOF.agents.base_agents.DataStreamAgent(name='', host=None, serializer=None, transport=None, attributes=None, backend=Backend.OSBRAIN, mesa_model=<class 'mesa.model.Model'>)

Able to simulate generation of datastream by loading a given DataStreamMET4FOF

Can be used in incremental training or batch training mode. To simulate batch training mode, set pretrain_size=-1 , otherwise, set pretrain_size and batch_size for the respective. See DataStreamMET4FOF on loading your own data set as a data stream.

agent_loop()

User defined method for the agent to execute for loop_wait seconds specified either in self.loop_wait or explicitly via init_agent_loop( loop_wait)

To start a new loop, call init_agent_loop(loop_wait) on the agent. Example of usage is to check the current_state of the agent and send data periodically.

init_parameters(stream=<agentMET4FOF.streams.base_streams.DataStreamMET4FOF object>, pretrain_size=None, batch_size=1, loop_wait=1, randomize=False)

Parameters:

• stream (DataStreamMET4FOF) – A DataStreamMET4FOF object which provides the sample data

• pretrain_size (int) – The number of sample data to send through in the first loop cycle, and subsequently, the batch_size will be used

• batch_size (int) – The number of sample data to send in every loop cycle

• loop_wait (int) – The duration to wait (seconds) at the end of each loop cycle before going into the next cycle

• randomize (bool) – Determines if the dataset should be shuffled before streaming

reset()

Reset the agent’s states and parameters

User can override this method to reset the specific parameters.

class agentMET4FOF.agents.base_agents.MonitorAgent(name='', host=None, serializer=None, transport=None, attributes=None, backend=Backend.OSBRAIN, mesa_model=<class 'mesa.model.Model'>)

Unique Agent for storing plots and data from messages received from input agents.

The dashboard searches for Monitor Agents’ buffer and plots to draw the graphs “plot” channel is used to receive base64 images from agents to plot on dashboard

plots

Dictionary of format {agent1_name : agent1_plot, agent2_name : agent2_plot}

Type:

dict

plot_filter

List of keys to filter the ‘data’ upon receiving message to be saved into memory Used to specifically select only a few keys to be plotted

Type:

list of str

custom_plot_function

a custom plot function that can be provided to handle the data in the monitor agents buffer (see AgentMET4FOF for details). The function gets provided with the content (value) of the buffer and with the string of the sender agent’s name as stored in the buffer’s keys. Additionally any other parameters can be provided as a dict in custom_plot_parameters.

Type:

callable

custom_plot_parameters

a custom dictionary of parameters that shall be provided to each call of the custom_plot_function

Type:

dict

init_parameters(plot_filter: List[str] | None = None, custom_plot_function: Callable[[...], Scatter] | None = None, **kwargs)

Initialize the monitor agent’s parameters

Parameters:

• plot_filter (list of str, optional) – List of keys to filter the ‘data’ upon receiving message to be saved into memory. Used to specifically select only a few keys to be plotted

• custom_plot_function (callable, optional) – a custom plot function that can be provided to handle the data in the monitor agents buffer (see AgentMET4FOF for details). The function gets provided with the content (value) of the buffer and with the string of the sender agent’s name as stored in the buffer’s keys. Additionally any other parameters can be provided as a dict in custom_plot_parameters. By default the data gets plotted as shown in the various tutorials.

• kwargs (Any) – custom key word parameters that shall be provided to each call of the custom_plot_function

on_received_message(message)

Handles incoming data from ‘default’ and ‘plot’ channels.

Stores ‘default’ data into buffer and ‘plot’ data into plots

Parameters:

message (dict) – Acceptable channel values are ‘default’ or ‘plot’

reset()

Reset the agent’s states and parameters

User can override this method to reset the specific parameters.

update_plot_memory(message: Dict[str, Any])

Updates plot figures stored in self.plots with the received message

Parameters:

message (dict) – Standard message format specified by AgentMET4FOF class Message[‘data’] needs to be base64 image string and can be nested in dictionary for multiple plots. Only the latest plot will be shown kept and does not keep a history of the plots.

Signal agents

class agentMET4FOF.agents.signal_agents.NoiseAgent(name='', host=None, serializer=None, transport=None, attributes=None, backend=Backend.OSBRAIN, mesa_model=<class 'mesa.model.Model'>)

An agent adding white noise to the incoming signal

init_parameters(noise_std: float | None = 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

property noise_std

Standard deviation of the distribution to randomly draw noise from

on_received_message(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>"
    }

class agentMET4FOF.agents.signal_agents.SineGeneratorAgent(name='', host=None, serializer=None, transport=None, attributes=None, backend=Backend.OSBRAIN, mesa_model=<class 'mesa.model.Model'>)

An agent streaming a sine signal

Takes samples from the SineGenerator and pushes them sample by sample to connected agents via its output channel.

agent_loop()

Model the agent’s behaviour

On state Running the agent will extract sample by sample the input data streams content and push it via invoking AgentMET4FOF.send_output().

init_parameters(sfreq=100, sine_freq=6.283185307179586, amplitude=1.0, initial_phase=0.0)

Initialize the input data

Initialize the input data stream as an instance of the SineGenerator class.

Parameters:

• sfreq (int) – sampling frequency for the underlying signal

• sine_freq (float) – frequency of the generated sine wave

• amplitude (float) – amplitude of the generated sine wave

• initial_phase (float) – initial phase (at t=0) of the generated sine wave

class agentMET4FOF.agents.signal_agents.SineWithJitterGeneratorAgent(name='', host=None, serializer=None, transport=None, attributes=None, backend=Backend.OSBRAIN, mesa_model=<class 'mesa.model.Model'>)

An agent streaming a sine signal

Takes samples from the SineWithJitterGenerator and pushes them sample by sample to connected agents via its output channel.

agent_loop()

Model the agent’s behaviour

On state Running the agent will extract sample by sample the input data streams content and push it via invoking AgentMET4FOF.send_output().

init_parameters(sfreq: int | None = 10, sine_freq: float | None = 0.15915494309189535, amplitude: float | None = 1.0, initial_phase: float | None = 0.0, jitter_std: float | None = 0.02)

Initialize the input data

Initialize the input data stream as an instance of the SineWithJitterGenerator class.

Parameters:

• sfreq (int, optional) – sampling frequency which determines the time step when next_sample() is called, defaults to 10

• sine_freq (float, optional) – frequency of the generated sine wave, defaults to \(\frac{1}{2 \pi}\)

• amplitude (float, optional) – amplitude of the generated sine wave, defaults to 1.0

• initial_phase (float, optional) – initial phase (at t=0) of the generated sine wave, defaults to 0.0

• jitter_std (float, optional) – the standard deviation of the distribution to randomly draw jitter from, defaults to 0.02

class agentMET4FOF.agents.signal_agents.StaticSineWithJitterGeneratorAgent(name='', host=None, serializer=None, transport=None, attributes=None, backend=Backend.OSBRAIN, mesa_model=<class 'mesa.model.Model'>)

An agent streaming a pre generated sine signal of fixed length with jitter

Takes samples from the StaticSineGeneratorWithJitter and pushes them sample by sample to connected agents via its output channel.

agent_loop()

Extract sample by sample the input data stream’s content and push it

init_parameters(num_cycles: int | None = 1000, jitter_std: float | None = 0.02)

Initialize the pre generated sine signal of fixed length with jitter

Initialize the static input data as an instance of the StaticSineWithJitterGenerator class with the provided parameters.

Parameters:

• num_cycles (int, optional) – numbers of cycles, determines the signal length by \(\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

An agent to reduce noise and jitter in a signal

class agentMET4FOF.agents.noise_jitter_removal_agents.MCMCMHNJ(fs, ydata, N, niter, tol, m0w, s0w, m0t, s0t, Mc, M0, Nc, Q)

This is the main class that implements the Bayesian Noise and jitter reduction

It is used by the NoiseJitterRemovalAgent.

MCMC used to determine the noise and jitter variances. Noise and jitter variances are then used in an iterative algorithm to remove the noise and jitter from the signal.

For a more detailed description of the individual methods are provided in a wiki

AnalyseSignalN()

Analyse signal to remove noise and jitter providing signal estimates

Associated uncertainty. Uses normalised independent variable.

NJAlgorithm(datax, data_y, p0, p0x)

Noise and Jitter Removal Algorithm

Iterative scheme that preprocesses data to reduce the effects of noise and jitter, resulting in an estimate of the true signal along with its associated uncertainty.

References

• Jagan et al. [Jagan2020]

static fgh_cubic(alpha, t)

Cubic function and its first and second derivative

KJ, LRW, PMH, Anupam Prasad Vedurmudi, Björn Ludwig Version 2020-04-22

Parameters:

• alpha – (4,N) Alpha parameters

• t – (m,1) Times

Returns:

• f – (m,N) Cubic function

• f1 – (m,N) Derivative of cubic

• f2 – (m,N) Second derivative of cubic

static jumprwg(A, L)

Jumping distribution for the Metropolis Hastings Gaussian random walk

KJ, LRW, PMH, Anupam Prasad Vedurmudi, Björn Ludwig Version 2020-04-22

Parameters:

• A(n – Samples at the current iteration

• N) – Samples at the current iteration

• L(n – Cholesky factor of variance of parameter vector.

• n) – Cholesky factor of variance of parameter vector.

Returns:

• As(n,N) – Proposed parameter array which is randomly sampled from the jumping distribution

• dp0 – The difference between the logarithm of the jumping distribution associated with moving from A(:,j) to As(:,j) and that associated with moving from As(:,j) to A(:,j), up to an additive constant. log P0(a|as) - log P0(as|a)

static ln_gauss_pdf_v(x, mu, sigma)

Log of the Gaussian pdf

KJ, LRW, PMH, Anupam Prasad Vedurmudi, Björn Ludwig Version 2020-03-12

Parameters:

• x – (m,1) Points at which pdf is to be evaluated

• mu – Mean of distribution

• sigma – Standard deviation of the distribution

Returns:

Log of the Gaussian pdf at x with mean mu and std sigma

Return type:

logf

static mcmcci(A, M0)

MCMC convergence indices for multiple chains

KJ, LRW, PMH, Anupam Prasad Vedurmudi, Björn Ludwig Version 2020-04-22

Parameters:

• A – (M, N) Chain samples, N chains of length M

• M0 – Length of burn-in, M > M0 >= 0.

Returns:

• Rhat – Convergence index. Rhat is expected to be greater than 1. The closer Rhat is to 1, the better the convergence.

• Neff – Estimate of the effective number of independent draws. Neff is expected to be less than (M-M0)*N.

• Note (If the calculated value of Rhat is < 1, then Rhat is set to 1 and Neff)

• set to (M-M0)*N, their limit values.

• Note (If N = 1 or M0 > M-2, Rhat = 0; Neff = 0.)

static mcmcmh(M, N, M0, Q, A0, tar, jump)

Metrolopolis-Hasting MCMC algorithm generating N chains of length M

For details about the algorithm please refer to: Gelman A, Carlin JB, Stern HS, Dunson DB, Vehtari A, Rubin DB. Bayesian data analysis. CRC press; 2013 Nov 1.

KJ, LRW, PMH, Anupam Prasad Vedurmudi, Björn Ludwig Version 2020-04-22

Parameters:

• M – Length of the chains

• N – Number of chains

• M0 – Burn in period

• Q – (nQ,1) Percentiles 0 <= Q(k) <= 100

• A0 – (n,N) Array of feasible starting points: the target distribution evaluated at A0(:,j) is strictly positive.

Returns:

• S(2+nQ,n) – Summary of A - mean, standard deviation and percentile limits, where the percentile limits are given by Q

• aP(N,1) – Acceptance percentages for AA calculated for each chain

• Rh(n,1) – Estimate of convergence. Theoretically Rh >= 1, and the closer to 1, the more evidence of convergence

• Ne(n,1) – Estimate of the number of effective number of independent draws

• AA(M,N,n) – Array storing the chains: A(i,j,k) is the kth element of the parameter vector stored as the ith member of the jth chain AA(1,j,:) = A0(:,j)

• IAA(M,N) – Acceptance indices. IAA(i,j) = 1 means that the proposal as(n,1) generated at the ith step of the jth chain was accepted so that AA(i,j,:) = as. IAA(i,j) = 0 means that the proposal as(n,1) generated at the ith step of the jth chain was rejected so that AA(i,j,:) = AA(i-1,j,:), i > 1. The first set of proposal coincide with A0 are all accepted, so IAA(1,j) = 1.

static mcsums(A, M0, Q)

Summary information from MC samples

KJ, LRW, PMH, Anupam Prasad Vedurmudi, Björn Ludwig Version 2020-04-22

Parameters:

• A – An (M,N) array that stores samples of size M x N

• M0 – Burn-in period with M > M0 >= 0

• Q – (nQ,1) Percentiles specifications, 0 <= Q(l) <= 100

Returns:

• abar(n,1) – Mean for each sample

• s(n,1) – Standard deviation for sample

• aQ(nQ,n) – Percentiles corresponding to Q

static tar_at(at, y, x, m0w, s0w, m0t, s0t)

Target dist for noise and jitter posterior dist

KJ, LRW, PMH, Anupam Prasad Vedurmudi, Björn Ludwig Version 2020-07-30

Parameters:

• at – (n+2,N) Parameters alpha, log(1/tau^2) and log(1/w^2)

• y – (m,1) Signal

• x – (m,1) time at which signal was recorded

• m0w – degree of belief in prior estimate for omega

• s0w – prior estimate of omega

• m0t – degree of belief in prior estimate for tau

• s0t – prior estimate of tau

Returns:

Log of the posterior distribution

Return type:

T

class agentMET4FOF.agents.noise_jitter_removal_agents.NoiseJitterRemovalAgent(name='', host=None, serializer=None, transport=None, attributes=None, backend=Backend.OSBRAIN, mesa_model=<class 'mesa.model.Model'>)

init_parameters(fs=100, ydata=array([], dtype=float64), N=15, niter=100, tol=1e-09, m0w=10, s0w=0.0005, m0t=10, s0t=0.0025, Mc=5000, M0=100, Nc=100, Q=50)

User provided function to initialize parameters of choice.

on_received_message(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.

Metrologically enabled base agents

class agentMET4FOF.agents.metrological_base_agents.MetrologicalAgent(name='', host=None, serializer=None, transport=None, attributes=None, backend=Backend.OSBRAIN, mesa_model=<class 'mesa.model.Model'>)

_input_data: Dict[str, Dict[str, TimeSeriesBuffer | Dict]]

Input dictionary of all incoming data including metadata:

dict like {
    <from>: {
        "buffer": TimeSeriesBuffer(maxlen=buffer_size),
        "metadata": MetaData(**kwargs).metadata,
    }

_output_data: Dict[str, Dict[str, TimeSeriesBuffer | MetaData]]

Output dictionary of all outgoing data including metadata:

dict like {
    <from>: {
        "buffer": TimeSeriesBuffer(maxlen=buffer_size),
        "metadata": MetaData(**kwargs).metadata,
    }

agent_loop()

User defined method for the agent to execute for loop_wait seconds specified either in self.loop_wait or explicitly via init_agent_loop( loop_wait)

To start a new loop, call init_agent_loop(loop_wait) on the agent. Example of usage is to check the current_state of the agent and send data periodically.

init_parameters(input_data_maxlen=25, output_data_maxlen=25)

User provided function to initialize parameters of choice.

on_received_message(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.

pack_data(data, channel='default')

Internal method to pack the data content into a dictionary before sending out.

Special case : if the data is already a message, then the from and senderType will be altered to this agent, without altering the data and channel within the message this is used for more succinct data processing and passing.

Parameters:

• data (argument) – Data content to be packed before sending out to agents.

• channel (str) – Key of dictionary which stores data

Returns:

• Packed message data (dict of the form {‘from’:agent_name, ‘data’: data,)

• ’senderType’ (agent_class, ‘channel’:channel_name}.)

class agentMET4FOF.agents.metrological_base_agents.MetrologicalMonitorAgent(name='', host=None, serializer=None, transport=None, attributes=None, backend=Backend.OSBRAIN, mesa_model=<class 'mesa.model.Model'>)

init_parameters(*args, **kwargs)

User provided function to initialize parameters of choice.

on_received_message(message)

Handles incoming data from default and plot channels

Feeds default data into the buffer as a dictionary:

dict like {
    "data": message["data"],
    "metadata": message["metadata"],
}

and hands over ‘plot’ data to plot memory.

Parameters:

message (dict) – Acceptable channel values are default or plot

reset()

Reset the agent’s states and parameters

User can override this method to reset the specific parameters.

update_plot_memory(message)

Updates plot figures stored in self.plots with the received message

Parameters:

message (dict) – Standard message format specified by AgentMET4FOF class Message[‘data’] needs to be base64 image string and can be nested in dictionary for multiple plots Only the latest plot will be shown kept and does not keep a history of the plots.

A metrologically enabled agent to reduce redundancy

class agentMET4FOF.agents.metrological_redundancy_agents.RedundancyAgent(name='', host=None, serializer=None, transport=None, attributes=None, backend=Backend.OSBRAIN, mesa_model=<class 'mesa.model.Model'>)

This is the main Redundancy Agent class

Redundancy means that there is more than one way to derive the value of the measurand Y from the values of the sensor data X_i. Following main cases are considered in the agent:

• Redundant measurement of the measurand Y by independent sensors directly measuring Y

• Redundant measurement of the measurand Y by correlated sensors directly measuring Y

• Redundant measurement of the measurand Y by correlated sensors X_i indirectly measuring Y, with a linear relationship y = a + A * x between the vector x of sensor values and the vector y containing the various (redundant) estimates of the measurand Y, where a is a vector and A a matrix both of appropriate size.

Main calculations are performed in calc_lcs() and calc_lcss(). Usage of the RedundancyAgent is relatively straightforward. Note that all static functions have their own test functions illustrating their usage. Details of the different methods are presented in their respective docstrings.

Please refer to other sections in this documentation for more information. A scientific publication explaining the ideas behind this agent can be found in [Kok20201]. Related work can be found in [Kok20202].

The usage of the Redundancy Agent is illustrated with two examples contained in two tutorials.

References

• Kok and Harris [Kok20201]

• Kok and Harris [Kok20202]

agent_loop()

Model the agent’s behaviour

On state Running the agent will extract sample by sample the input data streams content and push it via invoking send_output.

calc_best_est_lin_sys(a_arr, a_arr2d, x_arr, vx_arr2d, problim)

Calculate the best estimate of a linear system y = a + A * x

Additionally determine if the inputs are consistent in view of problim.

Parameters:

• a_arr (np.ndarray of shape (n_estimates)) – vector a of linear system y = a + A * x

• a_arr2d (np.ndarray of shape (n_estimates, n_sensors)) – matrix A of linear system y = a + A * x

• x_arr (np.ndarray of shape (n_sensors)) – vector with sensor values, vector x of linear system y = a + A * x

• vx_arr2d (np.ndarray of shape (n_sensors, n_sensors)) – uncertainty matrix associated with vector x_arr

• problim (float) – probability limit used for consistency evaluation. Typically 0.95.

Returns:

• isconsist (bool) – indicator whether provided estimates are consistent in view of problim

• ybest (float) – best estimate

• uybest (float) – standard uncertainty of best estimate

• chi2obs (float) – observed chi-squared value

static calc_best_estimate(y_arr, vy_arr2d, problim)

Calculate the best estimate for a set of estimates with uncertainties

Additionally determine if the set of estimates are consistent using a provided limit probability.

Parameters:

• y_arr (np.ndarray of shape (n)) – vector of estimates of a measurand Y

• vy_arr2d (np.ndarray of shape (n, n)) – uncertainty matrix associated with y_arr

• problim (float) – probability limit used for assessing the consistency of the estimates. Typically, problim equals 0.95.

Returns:

• isconsist (bool) – indicator whether provided estimates are consistent in view of problim

• ybest (float) – best estimate of measurand

• uybest (float) – uncertainty associated with ybest

• chi2obs (float) – observed value of chi-squared, used for consistency evaluation

static calc_consistent_estimates_no_corr(y_arr2d, uy_arr2d, prob_lim)

Calculation of consistent estimate for sets of estimates y_ij

The y_ij (contained in y_arr2d) are the elements of Y, where each set contains n_estims estimates. The uncertainties are assumed to be independent and given in uy_arr2d. The consistency test is using limit probability limit prob_lim. For each set of estimates, the best estimate, uncertainty, observed chi-2 value and a flag if the provided estimates were consistent given the model are given as output.

Parameters:

• y_arr2d (np.ndarray of size (n_rows, n_estimates)) – each row contains m=n_estimates independent estimates of a measurand

• uy_arr2d (np.ndarray of size (n_rows, n_estimates)) – each row contains the standard uncertainty u(y_ij) of y_ij = y_arr2d[i,j]

• prob_lim (float) – limit probability used in consistency test. Typically 0.95.

Returns:

• isconsist_arr (bool array of shape (n_rows)) – indicates for each row if the n_estimates are consistent or not

• ybest_arr (float or np.ndarray of float in shape (n_rows)) – contains the best estimate for each row of individual estimates

• uybest_arr (float or np.ndarray of float in shape (n_rows)) – contains the uncertainty associated with each best estimate for each row of y_arr2d

• chi2obs_arr (np.ndarray of float in shape (n_rows)) – observed chi-squared value for each row

calc_lcs(y_arr, vy_arr2d, problim)

Calculate the best estimate of a measurand with associated uncertainty matrix

Parameters:

• y_arr (np.ndarray of shape (n)) – vector with estimates of the measurand

• vy_arr2d (np.ndarray of shape (n, n)) – uncertainty matrix of the vector y_arr

• problim (float) – limit probability used in the consistency evaluation. Typically 0.95.

Returns:

• n_solutions (int or np.ndarray of ints) – number of solutions

• ybest (float or np.ndarray of floats) – best estimate

• uybest (float or np.ndarray of floats) – standard uncertainty of best estimate

• chi2obs (float or np.ndarray of floats) – observed chi-squared value

• indkeep (np.ndarray of shape (n) or (n_sols, n)) – indices of kept estimates

calc_lcss(a_arr, a_arr2d, x_arr, vx_arr2d, problim)

Calculation of the largest consistent subset of sensor values

Additionally the implied best estimate is returned.

Parameters:

• a_arr (np.ndarray of shape (n_estimates)) – vector a of linear system y = a + A * x

• a_arr2d (np.ndarray of shape (n_estimates, n_sensors)) – matrix A of linear system y = a + A * x

• x_arr (np.ndarray of shape (n_sensors)) – vector with sensor values vector x of linear system y = a + A * x

• vx_arr2d (np.ndarray of shape (n_sensors, n_sensors)) – uncertainty matrix associated with vector x_arr

• problim (float) – probability limit used for consistency evaluation. Typically 0.95

Returns:

• n_solutions (int or np.ndarray of ints) – number of solutions

• isconsist (bool or np.ndarray of bool) – indicator whether provided estimates are consistent in view of problim

• ybest (float or np.ndarray of floats) – best estimate

• uybest (float or np.ndarray of floats) – standard uncertainty of best estimate

• chi2obs (float or np.ndarray of floats) – observed chi-squared value

static get_combination(values, n_keep, certain_combinations_index)

Return a certain subset of n_keep elements in a given array

Parameters:

• values (np.ndarray) – original values

• n_keep (int) – number of elements in subset

• certain_combinations_index (int) – the index of the desired combination as a result of a call of combinations(values, n_keep)

static ind_reduce_a(a_arr2d, epszero)

Returns the index of a linear dependent row of a matrix A

The motivation for this is, that this row does not contribute any new information to the system.

Parameters:

• a_arr2d (np.ndarray) – The matrix to be reduced as 2-dimensional array

• epszero (float) – some small constant used for checking equality to zero

Returns:

the index of the last row that can be taken out

Return type:

int

init_lcss_parameters(fsam, f1, f2, ampl_ratio, phi1, phi2)

Additional parameters used for this particular example

Provides the prior knowledge needed to make the information contained in the data redundant. This method sets up the vector a and matrix A for the system y = a + A * x.

Parameters:

• fsam (float) – sampling frequency

• f1 (float) – first frequency of interest in signal

• f2 (float) – second frequency of interest in signal

• ampl_ratio (float) – ratio of the amplitudes of the two frequency components

• phi1 (float) – initial phase of first frequency component

• phi2 (float) – initial phase of second frequency component

init_parameters(input_data_maxlen: int = 25, output_data_maxlen: int = 25, sensor_key_list: list = None, n_pr: int = 1, problim: float = 0.9, calc_type: str = 'lcs')

Initialize the redundancy agent

Parameters:

• input_data_maxlen (int, optional) – Defaults to 25

• output_data_maxlen (int, optional) – Defaults to 25

• sensor_key_list (list of str, optional) – list containing the names of the sensors that should feed data to the Redundancy Agent. Defaults to None

• n_pr (int, optional) – size of the batch of data that is handled at a time by the Redundancy Agent. Defaults to 1

• problim (float, optional) – limit probability used for consistency evaluation. Defaults to .9

• calc_type (str, optional) – calculation type: ‘lcs’ or ‘lcss’. Defaults to ‘lcs’

on_received_message(message)

Handle incoming data from ‘default’ channels

Store ‘default’ data into an internal buffer.

Parameters:

message (dict) – Only acceptable channel value is ‘default’.

static print_input_lcss(x_arr, vx_arr2d, a_arr, a_arr2d, problim)

Prints the input parameters of the method

Parameters:

• x_arr (np.ndarray of shape (n_sensors)) – vector with sensor values, vector x of linear system y = a + A * x

• vx_arr2d (np.ndarray of shape (n_sensors, n_sensors)) – uncertainty matrix associated with vector x_arr

• a_arr (np.ndarray of shape (n_estimates)) – vector a of linear system y = a + A * x

• a_arr2d (np.ndarray of shape (n_estimates, n_sensors)) – matrix A of linear system y = a + A * x

• problim (float) – probability limit used for consistency evaluation. Typically 0.95

print_output_cbe(isconsist_arr, ybest_arr, uybest_arr, chi2obs_arr)

Function to print the full output of calc_best_estimate.

Parameters:

• isconsist_arr (bool array of shape (n_rows)) – indicates for each row if the n_estimates are consistent or not

• ybest_arr (np.ndarray of floats in shape (n_rows)) – contains the best estimate for each row of individual estimates

• uybest_arr (np.ndarray of floats in shape (n_rows)) – contains the uncertainty associated with each best estimate for each row of y_arr2d

• chi2obs_arr (np.ndarray of floats in shape (n_rows)) – observed chi-squared value for each row

static print_output_lcs(n_solutions, ybest, uybest, chi2obs, indkeep, y_arr)

Method to print the output of the method calc_lcs()

Parameters:

• n_solutions (int) – number of best solutions

• ybest (float or np.ndarray of shape (n_sols)) – best estimate or vector of best estimates

• uybest (float or np.ndarray of shape (n_sols)) – standard uncertainty of best estimate or vector with standard uncertainty of best estimates

• chi2obs (float) – observed chi-squared value of all best solutions

• indkeep (np.ndarray of shape (n) or (n_sols, n)) – indices of retained estimates of y_arr for the calculation of the best estimate ybest

• y_arr (np.ndarray of shape (n)) – individual estimates of measurand

static print_output_lcss(n_solutions, ybest, uybest, chi2obs, indkeep, x_arr, a_arr2d)

Prints the outputs of the method calc_lcss()

Parameters:

• n_solutions (int or np.ndarray of ints) – number of solutions

• ybest (float or np.ndarray of floats) – best estimate

• uybest (float or np.ndarray of floats) – standard uncertainty of best estimate

• chi2obs (float or np.ndarray of floats) – observed chi-squared value

• indkeep (np.ndarray of int) – indices of kept estimates

• x_arr (np.ndarray of shape (n_estimates)) – vector a of linear system y = a + A * x

• a_arr2d (np.ndarray of shape (n_estimates, n_sensors)) – matrix A of linear system y = a + A * x

Return type:

None

static print_output_single(isconsist, ybest, uybest, chi2obs)

Print the output of a single row of the calculate_best_estimate function

Parameters:

• isconsist (bool) – Indicates if provided estimates were consistent

• ybest (float) – best estimate

• uybest (float) – uncertainty of best estimate

• chi2obs (float) – observed value of chi-squared

static reduce_vx(x_arr, vx_arr2d, a_arr, a_arr2d, epszero)

Reduce the system if matrix Vx is not of full rank

This might be ambiguous, as constant sensor values or offsets have to be estimated and are not known.

Parameters:

• x_arr (np.ndarray) – The vector x to be reduced

• vx_arr2d (np.ndarray) – The matrix Vx to be reduced as 2-dimensional array

• a_arr (np.ndarray) – The vector a to be reduced

• a_arr2d (np.ndarray) – The matrix A to be reduced as 2-dimensional array

• epszero (float) – some small constant used for checking equality to zero

Returns:

The reduced matrices and vectors xred_arr, vxred_arr2d, ared_arr, ared_arr2d

Return type:

np.ndarray, np.ndarray, np.ndarray, np.ndarray

Metrologically enabled signal agents

class agentMET4FOF.agents.metrological_signal_agents.MetrologicalGeneratorAgent(name='', host=None, serializer=None, transport=None, attributes=None, backend=Backend.OSBRAIN, mesa_model=<class 'mesa.model.Model'>)

An agent streaming a specified signal

Takes samples from an instance of MetrologicalDataStreamMET4FOF with sampling frequency sfreq and signal frequency sine_freq and pushes them sample by sample to connected agents via its output channel.

agent_loop()

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.

init_parameters(signal: ~agentMET4FOF.streams.metrological_base_streams.MetrologicalDataStreamMET4FOF = <agentMET4FOF.streams.metrological_signal_streams.MetrologicalSineGenerator object>, **kwargs)

Initialize the input data stream

Parameters:

signal (MetrologicalDataStreamMET4FOF) – the underlying signal for the generator (defaults to MetrologicalSineGenerator)