sqlaggregator.aggregator module

class sqlaggregator.aggregator.SQLAggregateHistorian(config_path, **kwargs)[source]

Bases: volttron.platform.agent.base_aggregate_historian.AggregateHistorian

Agent to aggregate data in historian based on a specific time period. This aggregate historian aggregates data collected by SQLHistorian.

collect_aggregate(topic_ids, agg_type, start_time, end_time)[source]

Collect the aggregate data by querying the historian’s data store

  • topic_ids – list of topic ids for which aggregation should be performed.

  • agg_type – type of aggregation

  • start_time – start time for query (inclusive)

  • end_time – end time for query (exclusive)


a tuple of (aggregated value, count of record over which

this aggregation was computed)

configure(config_name, action, config)[source]

Converts aggregation time period into seconds, validates configuration values and calls the collect aggregate method for the first time

  • config_name – name of the config entry in store. We only use one config store entry with the default name config

  • action – “NEW or “UPDATE” code treats both the same way

  • config – configuration as json object


Query the aggregate_topics table and create a map of (topic name, aggregation type, aggregation time period) to topic id. This should be done as part of init


Returns a list of topic_map containing

{(agg_topic_name.lower(), agg_type, agg_time_period) :id}

Returns a list of supported aggregations


list of supported aggregations


Query the topics table and create a map of topic name to topic id. This should be done as part of init


Returns a list of topic_map containing {topic_name.lower():id}

initialize_aggregate_store(aggregation_topic_name, agg_type, agg_time_period, topics_meta)[source]

Create the data structure (table or collection) that is going to store the aggregate data for the give aggregation type and aggregation time period

  • aggregation_topic_name – Unique topic name for this aggregation. If aggregation is done over multiple points it is a unique name given by user, else it is same as topic_name for which aggregation is done

  • agg_type – The type of aggregation. For example, avg, sum etc.

  • agg_time_period – The time period of aggregation

  • topics_meta – String that represents the list of topics across which this aggregation is computed. It could be topic name pattern or list of topics. This information should go into metadata table


Return a aggregation_topic_id after inserting aggregation_topic_name into topics table

insert_aggregate(topic_id, agg_type, period, end_time, value, topic_ids)[source]

Insert aggregate data collected for a specific time period into database. Data is inserted into <agg_type>_<period> table

  • agg_topic_id – If len(topic_ids) is 1. This would be the same as the topic_ids[0]. Else this id corresponds to the unique topic name given by user for this aggregation across multiple points.

  • agg_type – type of aggregation

  • agg_time_period – The time period of aggregation

  • end_time – end time used for query records that got aggregated

  • topic_ids – topic ids for which aggregation was computed

  • value – aggregation result

update_aggregate_metadata(agg_id, aggregation_topic_name, topic_meta)[source]

Update aggregation_topic_name and topic_meta data for the given agg_id.

  • agg_id – Aggregation topic id for which update should be done

  • aggregation_topic_name – New aggregation_topic_name

  • topic_meta – new topic metadata

sqlaggregator.aggregator.main(argv=['/home/docs/checkouts/readthedocs.org/user_builds/volttron/envs/releases-8.x/lib/python3.6/site-packages/sphinx/__main__.py', '-T', '-E', '-b', 'html', '-d', '_build/doctrees', '-D', 'language=en', '.', '_build/html'])[source]

Main method called by the eggsecutable.