flexmeasures.data.models.reporting.aggregator
Classes
- class flexmeasures.data.models.reporting.aggregator.AggregatorReporter(config: dict | None = None, save_config=True, save_parameters=False, **kwargs)
Aggregate the sensors of a device group, or sensors named one by one.
A site’s flex-config already describes its topology: which devices sit behind which piece of equipment (their group), which sensor records each of them, which sign means consumption or production, and which sensor a group’s aggregate belongs on. Naming a group reports the measured aggregate of that group, so the report needs no topology of its own and cannot drift from the one the scheduler uses. A members filter narrows a group down to a category of its devices, such as every PV installation behind a connection.
Sensors can still be named one by one as input parameters, which is the way to aggregate something other than a site’s power, or to subtract one sensor’s data from another’s using weights.
Values are converted to the unit of the output sensor at each sensor’s own resolution, before being resampled to the output’s resolution, because converting between a stock and a flow (kWh to MW, say) depends on the resolution of the data being converted.
- _assign_sensors_and_source(results: list[dict[str, Any]]) list[dict[str, Any]]
Assign each result its sensor, and keep the sources the report chose per event.
A report over a group records an event that some members did not report under a source of its own, which assigning the report’s own source to every event would undo.
- _cached_inventory: Any = None
The site’s classified flex-config, built once per reporter and reused.
- static _check_one_source_per_event(df, sensor: Sensor, source, input_description: dict) None
Refuse data that holds more than one belief about one event, or several sources across events unless the caller narrowed the sources down.
The search keeps one belief per event wherever several sources report it, but an input that asks for more than the most recent belief of a single source still gets several beliefs about one event. Different versions of one source are not several sources, and neither is data a source filter has already narrowed, which is why a filtered search is accepted here even when more than one source remains.
- _collect_input_descriptions(input: list[dict[str, Any]], output_sensor: Sensor) list[dict[str, Any]]
List what to read, combining the input parameters with the members of a configured group.
The input descriptions are copied, so that reading them does not consume the parameters the reporter was given. The output sensor is never aggregated into itself, which would fold each run’s own result into the next one.
- _complete_events_superseding_a_partial_one(output_df: BeliefsDataFrame, output_sensor: Sensor, start: datetime, end: datetime) DatetimeIndex
Return the starts of the complete events of this report that a partial aggregate recorded since the last complete one stands in for.
A belief that says the same as its source’s belief before it is normally not saved again. A complete aggregate computed again after a partial one was recorded may say just what the complete source said before, but leaving it out would leave the partial aggregate as the most recent belief, which every consumer of one belief per event would read. So such a complete aggregate is saved even if unchanged. Only a partial aggregate that this report’s belief would supersede counts, which is one of the same version recorded no later. A report on a sensor that never had a partial aggregate of this version costs only the lookup of its sources.
- _compute_report(start: datetime, end: datetime, output: list[dict[str, Any]] | None = None, input: list[dict[str, Any]] | None = None, resolution: timedelta | None = None, belief_time: datetime | None = None, belief_horizon: timedelta | None = None) list[dict[str, Any]]
Read every sensor, express it in the output sensor’s unit and resolution, and aggregate.
- _filter_members(members: list[GroupMember]) list[GroupMember]
Narrow the group’s members down with the members filter, if one is configured.
- _group_asset() GenericAsset | None
Return the asset (or the asset of the sensor) the configured group refers to.
- _group_entry() dict | None
Return the flex-model entry of the group itself, which says where its aggregate is recorded.
- _group_members() list[GroupMember]
Resolve the readings that make up the configured group’s aggregate.
Membership comes from the scheduler’s device inventory, so adding a device to a group adds it to this report, and a group of groups is resolved to the devices behind it rather than to the inner groups’ own output sensors.
- _group_output_sensor(raise_if_missing: bool = True) Sensor | None
Return the sensor the group’s aggregate is recorded on.
A group referenced by sensor records on that sensor. A group referenced by asset records on the production or consumption output sensor its own entry names.
- _group_sign() bool
Return whether positive values on the sensor the aggregate is recorded on mean consumption.
A group referenced by asset says so through the field its own entry records under: an aggregate under production is production-positive, one under consumption is consumption-positive. A group referenced by a sensor has no such field, so that sensor’s own convention decides, which is production-positive unless the sensor says otherwise.
- _inventory()
Classify the site’s flex-config the way a scheduler of that site would.
Going through the scheduler’s own device inventory is what keeps one description of the site: it resolves a group to its leaf devices, follows groups nested inside groups, finds members wherever they sit below the site, and classifies each device’s role and sign convention. The window does not affect membership, so a nominal one is enough when the report has not named one yet. The result is cached, because building it costs a query per asset.
- static _is_ancestor_of(asset: GenericAsset, other: GenericAsset) bool
Tell whether asset sits above other in the asset tree.
- _label_incomplete_events(frame: DataFrame, incomplete: Series, column_sensor_ids: dict[str, int], sources: Series) Series
Record each event that some members did not report under a source that names them, in its missing-members.
The complete aggregate keeps its own source, which says nothing about missing members, so a consumer can tell a partial aggregate from a complete one, and an aggregate recorded once its data is complete supersedes the partial one. Only members count: a gap in a sensor named in input besides the group is aggregated over, as reports over named sensors always have. Which members are missing can differ per event, so one report may record under several sources. A sensor that fails again later is recorded under the source it was recorded under before, as sources with the same attributes are shared.
- static _member_readings(device) list[GroupMember]
Return the readings of one member device: one per sensor its measurements are on.
A device that names its own power sensor is read from that sensor, because that is where its measurements are. Its consumption and production references are where a schedule is written, so reading those instead would leave a realized report with nothing for this device. An entry that names no power sensor of its own records its two directions separately, and then contributes both, each with the sign convention of the field it was named under.
- _name_invalid_entry(root: GenericAsset, start, end) str
Return a phrase naming the asset whose flex-model entry the site’s flex-config fails on, if one can be pinned down.
Only reached once the site’s flex-config has already failed, so the cost of probing each asset’s own entry is paid on the way to an error message rather than on every report.
- _record_membership(members: list[GroupMember]) None
Record which sensors the group resolved to, as part of the report’s data source.
Membership comes from the flex-config, so the set of aggregated devices can change without the report’s configuration changing. Recording the resolved membership on the data source traces each stored report to the devices it aggregated: adding a device to a group records the next report under a different source. Changes to the membership itself are noted in the group asset’s audit log when they are made.
- static _referenced_sensor(reference: Any) Sensor | None
Resolve a sensor reference, which may be a Sensor, a SensorReference, an id, or a {“sensor”: …} dict.
- static _resample(values: Series, resolution: timedelta, sensor_resolution: timedelta, unit: str) Series
Resample one sensor’s converted values to the resolution the report is recorded at.
Whether a quantity adds up or averages over a longer event depends on what it is: energy adds up, power averages, so the unit of the output sensor decides which. Going the other way, to a finer resolution, a power holds through the event it was recorded over, while an energy is divided over it.
- _resolve_output(output: list[dict[str, Any]]) list[dict[str, Any]]
Settle which sensor the aggregate is recorded on.
An output named in the parameters wins, so one report can be sent elsewhere. Otherwise a configured group’s own entry says where its aggregate belongs.
- _settle_gaps(frame: DataFrame, aggregated: Series, column_sensor_ids: dict[str, int]) tuple[Series, Series]
Settle what to record for events that not every sensor reported, and under which source to record each event.
- _site_root() GenericAsset
Return the top-level ancestor of the group’s asset.
A device may join a group from anywhere below the site, so the flex-config has to be gathered from the site root, which is the asset a scheduler of this site would be given.
- _to_output_unit(values: Series, sensor: Sensor, output_sensor: Sensor) Series
Convert one sensor’s values to the unit of the output sensor, at the sensor’s own resolution.
The resolution matters: converting between a stock and a flow (kWh to MW, say) divides by the duration of an event, so this has to happen before the values are resampled to the output’s resolution.
- static _warn_about_gaps(frame: DataFrame, incomplete: Series) None
Say which sensors left gaps, and how many events they cost, because a blank aggregate is otherwise hard to place.
- _window_start: datetime | None = None
The report’s window, used only so the inventory is built the way a scheduler of this site would build it.
- get_data_source_attributes() dict
Describe this report’s data source, naming the members it resolved to and the ones it is missing, also when the config is not saved.
Which members an aggregate covers, and which it leaves out, says what its data is, rather than how to compute it again, so it is part of the source even where the rest of the config is not saved, as with flexmeasures add report –no-save-config. Without it, a changed membership and a partial aggregate would both be recorded under the complete report’s source.
- class flexmeasures.data.models.reporting.aggregator.GroupMember(sensor: Sensor, asset: GenericAsset | None, consumption_is_positive: bool, source_filters: dict[str, Any] | None = None)
One reading of one device of a group, as the flex-config describes it.
A device that records its consumption and its production on two sensors contributes two of these. source_filters carries the provenance the flex-config states for the device, so that a report reads the same beliefs the scheduler would.