pswamp.streaming.mqtt_io

Classes

`MQTTConsumer`
`MQTTProducer`
`MQTT_IO`	Online input/output object to be used by applications.

Module Contents

class pswamp.streaming.mqtt_io.MQTTConsumer(topic, hostname, port)

client

last_msg = ''

output_queue

get_next()

on_message(client, userdata, message)

__iter__()

__next__()

class pswamp.streaming.mqtt_io.MQTTProducer(hostname, port)

client

publish(topic, msg)

send(*args, **kwargs)

flush(*args, **kwargs)

class pswamp.streaming.mqtt_io.MQTT_IO(mqtt_kwargs, input_topic='pmudata', output_topic=None, status_topic='application.status', command_topic='application.commands')

Online input/output object to be used by applications.

TODO: Remove t_start from arguments.

Parameters:

kafka_kwargs – Kwargs that will be used in kafka consumers and producers.
input_topic – Kafka topic for input data stream.
output_topic – Kafka topic for output data stream.
status_topic – Kafka topic for status messages.
t_start – Not used anymore.
command_topic – Kafka topic for application commands.

input_topic = 'pmudata'

mqtt_kwargs

command_topic = 'application.commands'

input_stream

output_topic = None

status_topic = 'application.status'

status_producer

command_consumer

get_sample_data_frame()

Get sample data frame from input (without affecting input stream)

Returns:: Data frame
Return type:: _type_

get_config_frame()

Get config frame

Returns:: Config frame
Return type:: _type_

get_next_data_frame()

Get next dataframe

Returns:: Data frame
Return type:: _type_

get_next_command()

Get command frame

Returns:: Command frame
Return type:: _type_

handle_result(result)

Handle result once analysis completes. If result is not None, the result will be sent to the output topic.

Parameters:: result (_type_) – Result

handle_output(topic, output)

handle_status(message)

Sends the satus message to the status topic.

Parameters:: message (_type_) – Status message

seek_relative_input_offset(n_samples)