RUFAS.report_generator module

RUFAS.report_generator.average_aggregator(data: list[float]) → float

Calculates the average of a list of numbers.

Parameters

datalist[float]

A list of numbers whose average is to be calculated.

Returns

float

The average of the input numbers.

RUFAS.report_generator.division_aggregator(data: list[float]) → float | None

Divides the first number in the list by each of the subsequent numbers.

Parameters

datalist[float]

A list of numbers for the division operation.

Returns

float

The result of dividing the first number by each subsequent number. Returns None if the list is empty or has only one element.

RUFAS.report_generator.product_aggregator(data: list[float]) → float

Returns the product of a list of numbers.

Parameters

datalist[float]

A list of numbers whose product is to be calculated.

Returns

float

The product of the input numbers. Returns 1 for an empty list.

RUFAS.report_generator.sd_aggregator(data: list[float]) → float

Calculates the standard deviation of a list of numbers.

Parameters

datalist[float]

A list of numbers whose standard deviation is to be calculated.

Returns

float

The standard deviation of the input numbers.

RUFAS.report_generator.sum_aggregator(data: list[float]) → float

Returns the sum of a list of numbers.

Parameters

datalist[float]

A list of numbers whose sum is to be calculated.

Returns

float

The sum of the input numbers.

RUFAS.report_generator.subtraction_aggregator(data: list[float]) → float | None

Subtracts each subsequent number in the list from the first number.

Parameters

datalist[float]

A list of numbers for the subtraction operation.

Returns

float

The result of subtracting each subsequent number from the first number. Returns None if the list is empty or has only one element.

class RUFAS.report_generator.ReportGenerator(time=None)

Bases: object

A class to generate reports based on filtered data and aggregation criteria and store them in a dictionary.

Attributes

reportsdict[str, dict[str, list[Any]]]

A dictionary containing the generated reports, with the report name as the key and the report data as the value.

timeRufasTime | None

A RufasTime object used to track the simulation time

__init__(time=None) → None

Initializes the ReportGenerator.

clear_reports() → None

Clears the reports generated by the ReportGenerator.

Returns

None

generate_report(filter_content: dict[str, Any], filtered_pool: dict[str, dict[str, list[Any]]]) → list[dict[str, str | dict[str, str]]]

Generates a report specified by the given filter content.

Parameters

filter_contentdict[str, Any]

A dictionary containing the configuration for the report, including details such as ‘name’, ‘filters’, ‘cross_references’, and aggregation instructions.

filtered_pooldict[str, dict[str, list[Any]]]

The data pool from which reports are generated.

Returns

list[dict[str, str | dict[str, str]]]

A list of log events.

_get_horizontal_first_value(filter_content: dict[str, Any]) → bool

Check if the ‘horizontal_first’ property (when present) in the report filter is a boolean. If not, raise an error. Return the value of ‘horizontal_first’ or False as the default value.

Parameters

filter_contentdict[str, Any]

A dictionary containing the configuration for the report, including details such as ‘name’, ‘filters’, ‘cross_references’, and aggregation instructions.

Returns

bool

The value of ‘horizontal_first’ in the report filter, or False if it is not present.

Raises

ValueError

If the value of ‘horizontal_first’ in the report filter is not a boolean.

_prepare_report_data_to_be_graphed(graph_data: dict[str, Any], filter_content: dict[str, Any], individual_report_name: str) → list[dict[str, str | dict[str, str]]]

Prepare and send aggregated report data to Graph Generator to be graphed. Parameters ———- graph_data : dict[str, Any]

The report data to be graphed.

filter_contentdict[str, Any]

A dictionary containing the configuration for the report, including details such as ‘name’, ‘filters’, ‘cross_references’, and aggregation instructions.

individual_report_namestr

The name of the report to be graphed.

Returns

list[dict[str, str | dict[str, str]]]

Returns the logs from GraphGenerator.generate_graph()

_ensure_unique_report_name_with_timestamp(report_name: str | None) → str

Make the given report name unique, if it is not, by appending a timestamp to it.

Notes

The rules for generating the name are as follows: 1. If the report name is None, the name is set to untitled_<timestamp>. 2. If the name is not a string, it is converted to a string. 3. If the name is already present in the list of previously generated reports, the timestamp is appended

to the name to make it unique.

4. Otherwise, return the report name as is.

Parameters

report_namestr | None

The name of the report.

Returns

str

The unique name for the report.

_check_for_missing_references(references: list[str]) → None

Checks if all the referenced reports are present in the list of previously generated reports.

Parameters

referenceslist[str]

The list of report references to check.

Raises

KeyError

If any of the report references are missing.

_get_reports_by_regex(regex_patterns: list[str]) → dict[str, dict[str, list[Any]]]

Retrieve reports based on matching the existing report names with the given regex patterns.

Notes

Each pattern is checked for a full match against the report names. A “full match” means that the regex must match the entire string of the report name from start to finish, without partial matches. This reduces the potential for false positives.

Parameters

regex_patternsList[str]

A list of regex patterns to match with the existing report names.

Returns

dict[str, dict[str, list[Any]]]

The matching reports.

_perform_aggregations(filtered_pool: dict[str, dict[str, list[Any]]], filter_content: dict[str, Any]) → tuple[dict[str, dict[str, list[Any]]] | dict[str, list[Any]], list[dict[str, str | dict[str, str]]]]

Fetches aggregation keys from the filter content and applies aggregation to the data.

Parameters

filtered_pooldict[str, dict[str, list[Any]]]

The data pool from which the report is to be generated, structured as a dictionary.

filter_contentdict[str, Any]

A dictionary containing filter criteria, aggregation instructions, and scalar operation details.

Returns

tuple[dict[str, dict[str, list[Any]]] | dict[str, list[Any]], list[dict[str, str | dict[str, str]]]]

If no aggregation is specified, all the columns will be returned. If both horizontal and vertical aggregations are specified, the returned dictionary will have one key

that is either “hor_ver_agg” or “ver_hor_agg” depending on the value of the “horizontal_first” key in the filter content.

If only horizontal aggregation is specified, the returned dictionary will have one key “hor_agg”. If only vertical aggregation is specified, the returned dictionary will have one key “ver_agg”. The second element is the event logs.

Raises

ValueError

If there is an error preparing the report data. If the report data is empty. If the type of horizontal or vertical aggregation is not supported.

_route_aggregator_functions(report_data: dict[str, dict[str, list[Any]]], filter_content: dict[str, Any], horizontal_agg_key: str | None = None, vertical_agg_key: str | None = None) → tuple[dict[str, dict[str, list[Any]]] | dict[str, list[Any]], list[dict[str, str | dict[str, str]]]]

Routes report data to appropriate vertical and horizontal aggregator functions.

Parameters

report_datadict[str, dict[str, list[Any]]]

The data to be aggregated.

filter_contentdict[str, Any]

The report filter settings.

horizontal_agg_keyOptional[str], optional

The horizontal aggregator function key.

vertical_agg_keyOptional[str], optional

The vertical aggregator function key.

Returns

tuple[dict[str, dict[str, list[Any]]] | dict[str, list[Any]], list[dict[str, str | dict[str, str]]]]

The aggregated data and the logs to be passed to Output Manager.

static _update_key(key: str) → str

Updates dictionary keys to keep units at the end of the key.

Parameters

keystr

The key to be updated

Returns

str

The updated key.

_aggregate_units(report_data: dict[str, list[float]], aggregator: Callable[[list[float]], float] | Callable[[list[float]], float | None], simplify_units: bool) → tuple[str | Any, dict[str, str | dict[str, str]]]

Creates the appropriate units for the associated aggregator function used.

Parameters

report_datadict[str, list[float]]

The data pool to be aggregated, structured as a dictionary of lists.

aggregatorCallable[[list[float]], float] | Callable[[list[float]], float | None]

The aggregation function to be used.

simplify_unitsbool

Whether to simplify and reduce the units.

Returns

tuple[str | Any, dict[str, str | dict[str, str]]]

The expected units of aggregating the report data using the accompanying aggregator function and the event logs.

static _combine_units(numerator1: dict[str, int], denominator1: dict[str, int], numerator2: dict[str, int], denominator2: dict[str, int], operation: str, simplify_units: bool) → tuple[dict[str, int], dict[str, int], dict[str, str | dict[str, str]]]

Combines two sets of units (numerator and denominator) based on the specified operation.

Parameters

numerator1dict[str, int]

First set of numerator units, where keys are unit names (str) and values are exponents (int).

denominator1dict[str, int]

First set of denominator units, where keys are unit names (str) and values are exponents (int).

numerator2dict[str, int]

Second set of numerator units, where keys are unit names (str) and values are exponents (int).

denominator2dict[str, int]

Second set of denominator units, where keys are unit names (str) and values are exponents (int).

operationstr

The operation to combine the units. Can be one of ‘product’, ‘division’, ‘sum’, ‘subtraction’, ‘average’, ‘SD’.

simplify_unitsbool

Whether to simplify and reduce the units.

Returns

tuple[dict[str, int], dict[str, int], dict[str, str | dict[str, str]]]

Two dictionaries representing the combined numerator and denominator units along with the event logs.

Raises

ValueError

If the operation is addition or subtraction and the units are not the same.

_handle_horizontal_and_vertical_aggregations(aggregate_report: dict[str, dict[str, list[Any]]] | dict[str, list[Any]], horizontal_agg_key: str, vertical_agg_key: str, filter_content: dict[str, Any]) → tuple[dict[str, list[Any]], list[dict[str, str | dict[str, str]]]]

Handles both horizontal and vertical aggregations on the report data.

Parameters

aggregate_reportdict[str, dict[str, list[Any]]] | dict[str, list[Any]]

The report data to be aggregated.

horizontal_agg_keystr

The key for the horizontal aggregation function.

vertical_agg_keystr

The key for the vertical aggregation function.

filter_contentdict[str, Any]

A dictionary containing filter criteria, aggregation instructions, and scalar operation details.

Returns

tuple[dict[str, list[Any]], list[dict[str, str | dict[str, str]]]]

The aggregated report data and the event log.

_extract_aggregation_keys(filter_content: dict[str, Any]) → tuple[str | None, str | None]

Extracts horizontal and vertical aggregation keys from the filter content and validates them against supported aggregation types.

Parameters

filter_contentdict[str, Any]

A dictionary containing filter criteria, aggregation instructions, and scalar operation details. It should include keys for ‘horizontal_aggregation’ and ‘vertical_aggregation’ if applicable.

Returns

tuple[str | None, str | None]

A tuple containing the horizontal and vertical aggregation keys, respectively. Returns None for each key if it is not specified or if it is not among the supported types.

Raises

ValueError

Raised if the specified horizontal or vertical aggregation type is not supported. Supported types are defined in the AGGREGATION_FUNCTIONS dictionary.

_apply_horizontal_aggregation(report_data: dict[str, list[float]], loop_list: list[str], aggregator: Callable[[list[float]], float] | Callable[[list[float]], float | None], simplify_units: bool) → tuple[list[float], str, list[dict[str, str | dict[str, str]]]]

Performs horizontal aggregation on report data using a specified aggregator function.

Parameters

report_datadict[str, list[float]]

The data pool to be aggregated, structured as a dictionary of lists.

loop_listlist[str]

List of keys indicating the order in which to aggregate data.

aggregatorCallable[[list[float]], float] | Callable[[list[float]], float | None]

The aggregation function to be used.

simplify_unitsbool

Whether to reduce the units.

Returns

tuple[list[float], str, list[dict[str, str | dict[str, str]]]]

The horizontally aggregated data as a list, the aggregated units, and the event logs to be passed to Output Manager.

Raises

ValueError

If the data to be aggregated has different lengths.

_apply_vertical_aggregation(report_data: dict[str, dict[str, list[float | None]]] | dict[str, list[float | None]], aggregator: Callable[[list[float]], float] | Callable[[list[float]], float | None]) → tuple[dict[str, list[float | None]], list[dict[str, str | dict[str, str]]]]

Performs vertical aggregation on report data using a specified aggregator function.

Parameters

report_datadict[str, dict[str, list[float | None]]] | dict[str, list[float | None]]

The data pool to be aggregated, structured as a dictionary of lists or a dictionary of dictionaries of lists.

aggregatorCallable[[list[float]], float] | Callable[[list[float]], float | None]

The aggregation function to be used.

Returns

tuple[dict[str, list[float | None]], list[dict[str, str | dict[str, str]]]]

The vertically aggregated data as a dictionary of lists and the logs to be passed to OutputManager.

_handle_aggregation(aggregator: Callable[[list[float]], float] | Callable[[list[float]], float | None], data: list[float], key: str) → tuple[float | None, dict[str, str | dict[str, str]]]

Wrapper function for RG aggregators to catch Nones and NaNs.

Parameters

aggregatorCallable[[list[float]], float] | Callable[[list[float]], float | None]

The aggregator function being called from the report filter.

datalist[float]

The data to be aggregated.

keystr

The key for the data being aggregated.

Returns

tuple[float | None, dict[str, str | dict[str, str]]]

The resulting aggregated data and the aggregation logs to be returned to OutputManager. Returns None and an error message if the data contains any values that cannot be aggregated.

_add_constants_to_report_data(report_data: dict[str, list[float]], filter_content: dict[str, Any]) → list[dict[str, str | dict[str, str]]]

Add constants to the report data.

Examples

An example of a configuration for constants:

“constants”: {

“Kilograms to Pounds”: 2.20462, “Pounds to Dollars”: 10

}

Parameters

report_datadict[str, list[float]]

The data to which constants need to be added.

filter_contentdict[str, Any]

A dictionary containing filter criteria, aggregation instructions, and scalar operation details.

Returns

list[dict[str, str | dict[str, str]]]

A list of warnings, logs, and errors to be returned to Output Manager for logging.

Raises

ValueError

If the name or value of any constant is not valid.

_add_units_to_constants(constants_config: dict[str, int | float]) → tuple[dict[str, int | float], list[dict[str, str | dict[str, str]]]]

Checks constants provided in filter file against GeneralConstants and adds appropriate measurement units.

Parameters

constants_configdict[str, int | float]

A dictionary containing the names and values of the constants to be added to the report data.

Returns

tuple[dict[str, int | float], list[dict[str, str | dict[str, str]]]]

A tuple of the dictionary containing the names and values of the constants to be added to the report data and any warnings to be passed to Output Manager for logging.

_validate_constants(existing_reports: dict[str, list[Any]], constants_config: dict[str, int | float]) → None

Validates the names and values of the constants in the constants_config.

This method checks that each constant’s name is not None, a non-empty string, and is unique compared to the names of all the existing reports. It also verifies that each constant’s value is a number (int or float) and not None.

Parameters

existing_reportsdict[str, list[Any]]

The dictionary containing the names and values of the reports that have already been generated.

constants_configdict[str, int | float]

A dictionary containing the names and values of the constants to be added to the report data.

Raises

ValueError

If a constant name already exists in the report data. If a constant name or value is None. If a constant name is not a string. If a constant name is empty. If a constant value is not a number.

static _normalize_constant_name(name: str) → str

Normalize the constant name by converting to lowercase and removing underscores and spaces.

static _add_var_units(report_data: dict[str, dict[str, list[Any]]]) → dict[str, dict[str, list[Any]]]

Adds variable units to variable name.

Parameters

report_datadict[str, dict[str, list[Any]]]

The data to be reported.

Returns

dict[str, dict[str, list[Any]]]

The updated data with units added.