pswamp.streaming
Submodules
Classes
Wrapper for Kafka producer. |
|
Wrapper for Kafka consumer. |
|
Online input/output object to be used by applications. |
|
Online input/output object to be used by applications. |
|
Package Contents
- class pswamp.streaming.KafkaProducer(*args, use_nqkafka=False, **kwargs)
Wrapper for Kafka producer.
- Parameters:
use_nqkafka (bool, optional) – Determines whether self.instance is a NQKafkaProducer (from the nqkafka package, can be thought of as a “mock”) or actual KafkaProducer (from the kafka-python package), which connects to a Kafka server. Defaults to False.
*args – Variable length argument list.
**kwargs – Arbitrary keyword arguments.
- instance
Determined from input parameters.
- Type:
nqkafka.NQKafkaProducerorkafka.KafkaProducer
- instance
- __getattr__(name)
Inherits the same attributes as the instance.
- Parameters:
name (_type_) – Name of the attribute.
- Returns:
The attribute of the instance.
- __iter__()
The iterator is the same as the iterator of the instance.
- Returns:
The instance.
- class pswamp.streaming.KafkaConsumer(*args, use_nqkafka=False, **kwargs)
Wrapper for Kafka consumer.
- Parameters:
use_nqkafka (bool, optional) – Determines whether self.instance is a NQKafkaConsumer (from the nqkafka package, can be thought of as a “mock”) or actual KafkaConsumer (from the kafka-python package), which connects to a Kafka server. Defaults to False.
*args – Variable length argument list.
**kwargs – Arbitrary keyword arguments.
- instance
Determined from input parameters.
- Type:
nqkafka.NQKafkaConsumerorkafka.KafkaConsumer
- instance
- __getattr__(name)
Inherits the same attributes as the instance.
- Parameters:
name (_type_) – Name of the attribute.
- Returns:
The attribute of the instance.
- __iter__()
The iterator is the same as the iterator of the instance.
- Returns:
The instance.
- class pswamp.streaming.KafkaIO(kafka_kwargs, input_topic='pmudata', output_topic=None, status_topic='application.status', t_start=None, 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'
- kafka_kwargs
- kafka_server
- 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_sample_pmu_data_frame()
Included for backwards compatibility
- 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)
Changes the offset of the input data consumer according to the specified number of samples.
- Parameters:
n_samples (int) – Number of samples to shift
- class pswamp.streaming.MQTTProducer(hostname, port)
- client
- publish(topic, msg)
- send(*args, **kwargs)
- flush(*args, **kwargs)
- class pswamp.streaming.MQTTConsumer(topic, hostname, port)
- client
- last_msg = ''
- output_queue
- get_next()
- on_message(client, userdata, message)
- __iter__()
- __next__()
- class pswamp.streaming.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)