VeraGridEngine.Simulations package

Subpackages

Submodules

VeraGridEngine.Simulations.driver_handler module

VeraGridEngine.Simulations.driver_handler.create_driver(grid: MultiCircuit, driver_tpe: SimulationTypes, time_indices: ndarray[tuple[Any, ...], dtype[int64]] | None) → AvailableTransferCapacityDriver | AvailableTransferCapacityTimeSeriesDriver | ContingencyAnalysisDriver | ContingencyAnalysisTimeSeriesDriver | ContinuationPowerFlowDriver | LinearAnalysisDriver | LinearAnalysisTimeSeriesDriver | OptimalPowerFlowDriver | OptimalPowerFlowTimeSeriesDriver | PowerFlowDriver | PowerFlowDriver3Ph | PowerFlowTimeSeriesDriver | StateEstimationDriver | ShortCircuitDriver | StochasticPowerFlowDriver | ClusteringDriver | CascadingDriver | SigmaAnalysisDriver | OptimalNetTransferCapacityDriver | OptimalNetTransferCapacityTimeSeriesDriver | NodeGroupsDriver | InputsAnalysisDriver | InvestmentsEvaluationDriver | CatalogueOptimizationDriver | NodalCapacityTimeSeriesDriver | ReliabilityStudyDriver | RmsSimulationDriver | SmallSignalStabilityRmsDriver | SmallSignalStabilityEmtDriver | EmtSimulationDriver | None[source]

Create driver with the results :param grid: MultiCircuit instance :param driver_tpe: SimulationTypes :param time_indices: list of time indices :return: Driver or None

VeraGridEngine.Simulations.driver_template module

class VeraGridEngine.Simulations.driver_template.DriverTemplate(grid: MultiCircuit, engine: EngineType = VeraGrid)[source]

Bases: object

Base driver template

cancel()[source]

Cancel the simulation

copy_signals(other: TimeSeriesDriverTemplate)[source]

Copy the signals from another driver :param other: :return:

done_signal
elapsed
engine
get_save_data() → DriverToSave[source]

Get save data representation of this driver :return: DriverToSave

get_steps()[source]

Get the number of steps in the simulation :return:

grid: MultiCircuit
isRunning()[source]
Returns:

is_cancel() → bool[source]

Check if cancel was activated :return:

logger
name = 'Template'
progress_signal
progress_text
report_done(txt='done!', val=0.0)[source]

Report done

report_progress(val: float)[source]

Report progress :param val: float value

report_progress2(current: int, total: int)[source]

Report progress :param current: current value (zero based) :param total: total value

report_text(val: str)[source]

Report text :param val: text value

results
run()[source]
tic(skip_logger=False)[source]

Register start of time

toc(skip_logger=False)[source]

Register end of time :param skip_logger: skip logging this?

tpe = 'Template'
class VeraGridEngine.Simulations.driver_template.DriverToSave(name: str, tpe: SimulationTypes, results: ResultsTemplate, logger: Logger)[source]

Bases: object

Wrapper to save a driver

logger: Logger
name
results: ResultsTemplate
tpe: SimulationTypes
class VeraGridEngine.Simulations.driver_template.DummySignal(tpe: type = <class 'str'>)[source]

Bases: object

Qt signal placeholder to not to import QT in the engine

connect(val)[source]
Parameters:

val

emit(val: str | float = '') → None[source]
tpe
class VeraGridEngine.Simulations.driver_template.TimeSeriesDriverTemplate(grid: MultiCircuit, time_indices: IntVec | None, clustering_results: ClusteringResults | None = None, engine: EngineType = VeraGrid, check_time_series: bool = True)[source]

Bases: DriverTemplate

Time series driver template

clustering_results: ClusteringResults | None
get_fuel_emissions_energy_calculations(gen_p: ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]], gen_cost: ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]])[source]

Calculate fuel emissions and energy cost :param gen_p: :param gen_cost: :return:

get_steps()[source]

Get time steps list of strings

original_sample_idx: IntVec
sampled_probabilities: Vec
time_indices: IntVec
using_clusters

VeraGridEngine.Simulations.options_template module

class VeraGridEngine.Simulations.options_template.OptionsTemplate(name: str)[source]

Bases: EditableDevice

Options template

CLASS_NON_EDITABLE_PROPERTIES: Tuple[str, ...] = ('idtag', 'diff_changes')
CLASS_PROPERTIES_WITH_PROFILE: Dict[str, str] = {}
CLASS_PROPERTY_DECLARATIONS: Tuple[GCProp, ...] = (prop:idtag, prop:name, prop:code, prop:rdfid, prop:action, prop:comment, prop:diff_changes)
CLASS_PROPERTY_LIST: Tuple[GCProp, ...] = (prop:idtag, prop:name, prop:code, prop:rdfid, prop:action, prop:comment, prop:diff_changes)
CLASS_REGISTERED_PROPERTIES: Dict[str, GCProp] = {'action': prop:action, 'code': prop:code, 'comment': prop:comment, 'diff_changes': prop:diff_changes, 'idtag': prop:idtag, 'name': prop:name, 'rdfid': prop:rdfid}
action: ActionType
comment: str
device_type: DeviceType
diff_changes
selected_to_merge

VeraGridEngine.Simulations.results_table module

class VeraGridEngine.Simulations.results_table.ResultsTable(data: ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]], columns: ndarray[tuple[Any, ...], dtype[str_]], index: ndarray[tuple[Any, ...], dtype[str_]] | DatetimeIndex, title: str, cols_device_type: DeviceType, idx_device_type: DeviceType, units: str = '', xlabel: str = '', ylabel: str = '', editable=False, palette=None, editable_min_idx: int = -1, decimals: int = 6)[source]

Bases: object

Class to populate a Qt table view with data from the results

c
property col_devices
Returns:

cols_c
cols_device_type: DeviceType
convert_to_abs()[source]

Convert the data to abs :return:

convert_to_cdf()[source]

Convert the data in-place to CDF based :return:

copy_to_column(row: int, col: int)[source]

Copies one value to all the column @param row: Row of the value @param col: Column of the value @return: Nothing

property data

Backward-compatible alias for the table numeric payload.

Returns:

Table data array.

data_c
decimals: int
editable
editable_min_idx
format_string
formatter
get_data()[source]

Returns: index, columns, data

get_data_frame()[source]

Save data to csv

idx_device_type: DeviceType
property idx_devices
Returns:

index_c
isDate
is_complex() → bool[source]

Is the data complex? :return:

palette
plot(ax=None, selected_col_idx=None, selected_rows=None, stacked=False)[source]

Plot the data model :param ax: Matplotlib axis :param selected_col_idx: list of selected column indices :param selected_rows: list of rows to plot :param stacked: Stack plot?

plot_device(ax=None, device_idx: int = 0, stacked=False, title: str = '')[source]

Plot the data model :param ax: Matplotlib axis :param device_idx: list of selected column indices :param stacked: Stack plot? :param title: Title of the plot

r
save_to_csv(file_name)[source]

Save data to csv :param file_name:

save_to_excel(file_name)[source]

save data to excel :param file_name:

search_in_columns(txt)[source]

Search stuff :param txt: :return:

search_in_rows(txt)[source]

Search stuff :param txt: :return:

set_col_devices(devices_list: List[Generator | Battery | Load | ExternalGrid | StaticGenerator | Shunt | ControllableShunt | CurrentInjection | Line | DcLine | Transformer2W | HvdcLine | VSC | UPFC | Winding | Switch | SeriesReactance | FluidNode | FluidPath | FluidP2x | FluidTurbine | FluidPump | Substation | Bus | BusBar | VoltageLevel | Country | Region | Community | Municipality | Area | Zone | Transformer3W | TransformerNW | OverheadLineType | Wire | TransformerType | EmissionGas | BranchGroup | LineLocations | LineLocation | ModellingAuthority | Facility | Fuel | Investment | InvestmentsGroup | Contingency | ContingencyGroup | RemedialAction | RemedialActionGroup | Technology | Owner | UndergroundLineType | SequenceLineType | RmsModelTemplate | EmtModelTemplate | FmuTemplate | RmsEvent | RmsEventsGroup | ShortCircuitEvent | IfMeasurement | ItMeasurement | QfMeasurement | PfMeasurement | QtMeasurement | PtMeasurement | QiMeasurement | PiMeasurement | VmMeasurement | VaMeasurement | PgMeasurement | QgMeasurement])[source]

Set the list of devices that matches the results for filtering :param devices_list:

set_idx_devices(devices_list: List[Generator | Battery | Load | ExternalGrid | StaticGenerator | Shunt | ControllableShunt | CurrentInjection | Line | DcLine | Transformer2W | HvdcLine | VSC | UPFC | Winding | Switch | SeriesReactance | FluidNode | FluidPath | FluidP2x | FluidTurbine | FluidPump | Substation | Bus | BusBar | VoltageLevel | Country | Region | Community | Municipality | Area | Zone | Transformer3W | TransformerNW | OverheadLineType | Wire | TransformerType | EmissionGas | BranchGroup | LineLocations | LineLocation | ModellingAuthority | Facility | Fuel | Investment | InvestmentsGroup | Contingency | ContingencyGroup | RemedialAction | RemedialActionGroup | Technology | Owner | UndergroundLineType | SequenceLineType | RmsModelTemplate | EmtModelTemplate | FmuTemplate | RmsEvent | RmsEventsGroup | ShortCircuitEvent | IfMeasurement | ItMeasurement | QfMeasurement | PfMeasurement | QtMeasurement | PtMeasurement | QiMeasurement | PiMeasurement | VmMeasurement | VaMeasurement | PgMeasurement | QgMeasurement])[source]

Set the list of devices that matches the results for filtering :param devices_list:

slice_all(row_idx, col_idx) → ResultsTable[source]

Make rows slicing :param row_idx: indices of the rows :param col_idx: indices of the columns :return: ResultsTable

slice_cols(col_idx) → ResultsTable[source]

Make column slicing :param col_idx: indices of the columns :return: Nothing

slice_rows(idx) → ResultsTable[source]

Make rows slicing :param idx: indices of the columns :return: Nothing

sort_column(c: int, max_to_min: bool = True)[source]
Parameters:

  • c

  • max_to_min

Returns:

title
to_df() → DataFrame[source]

get DataFrame

transpose()[source]

Transpose the results in-place

units
x_label
y_label

VeraGridEngine.Simulations.results_template module

class VeraGridEngine.Simulations.results_template.ResultsProperty[source]

Bases: object

expandable
name: str
old_names: Tuple[str, ...]
tpe: Any
class VeraGridEngine.Simulations.results_template.ResultsTemplate[source]

Bases: object

CLASS_DATA_VARIABLES = {}
CLASS_RESULTS_DECLARATIONS = ()
F: ndarray[tuple[Any, ...], dtype[int64]] | None
T: ndarray[tuple[Any, ...], dtype[int64]] | None
activate_plotting()[source]
Returns:

apply_new_rates(rates: ndarray[tuple[Any, ...], dtype[float64]])[source]
Parameters:

rates

apply_new_time_series_rates(rates: ndarray[tuple[Any, ...], dtype[float64]])[source]
Parameters:

rates

area_names: ndarray[tuple[Any, ...], dtype[str_]] | None
available_results: Dict[ResultTypes, List[ResultTypes]]
bus_area_indices: ndarray[tuple[Any, ...], dtype[int64]] | None
clustering_results
consolidate_after_loading()[source]

Consolidate

property data_variables: Dict[str, ResultsProperty]
Returns:

deactivate_plotting()[source]
Returns:

expand_clustered_results()[source]

Expand all arrays using the clustering info

fill_circuit_info(grid: MultiCircuit)[source]
Parameters:

grid

Returns:

fill_simulation_info(grid: MultiCircuit)[source]
Parameters:

grid

Returns:

get_branch_values_per_area(branch_values: ndarray[tuple[Any, ...], dtype[float64]], area_names: ndarray[tuple[Any, ...], dtype[str_]], bus_area_indices: ndarray[tuple[Any, ...], dtype[int64]], F: ndarray[tuple[Any, ...], dtype[int64]], T: ndarray[tuple[Any, ...], dtype[int64]])[source]

Split array of branch-related values per area :param branch_values: :param area_names: :param bus_area_indices: :param F: :param T: :return:

static get_bus_values_per_area(bus_values: ndarray[tuple[Any, ...], dtype[float64]], area_names: ndarray[tuple[Any, ...], dtype[str_]], bus_area_indices: ndarray[tuple[Any, ...], dtype[int64]]) → ndarray[tuple[Any, ...], dtype[float64]][source]

Split array of bus-related values per area :param bus_values: :param area_names: :param bus_area_indices: :return:

get_dict() → Dict[str, Any][source]

Get data to pass via json :return:

get_hvdc_values_per_area(hvdc_values: ndarray[tuple[Any, ...], dtype[float64]], area_names: ndarray[tuple[Any, ...], dtype[str_]], bus_area_indices: ndarray[tuple[Any, ...], dtype[int64]], hvdc_F: ndarray[tuple[Any, ...], dtype[int64]], hvdc_T: ndarray[tuple[Any, ...], dtype[int64]])[source]

Split array of hvdc-related values per area :param hvdc_values: :param area_names: :param bus_area_indices: :param hvdc_F: :param hvdc_T: :return:

get_inter_area_flows(area_names: ndarray[tuple[Any, ...], dtype[str_]], F: ndarray[tuple[Any, ...], dtype[int64]], T: ndarray[tuple[Any, ...], dtype[int64]], Sf: ndarray[tuple[Any, ...], dtype[complex128]], hvdc_F: ndarray[tuple[Any, ...], dtype[int64]], hvdc_T: ndarray[tuple[Any, ...], dtype[int64]], hvdc_Pf: ndarray[tuple[Any, ...], dtype[float64]], bus_area_indices: ndarray[tuple[Any, ...], dtype[int64]]) → ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]][source]
Parameters:

  • area_names

  • F

  • T

  • Sf

  • hvdc_F

  • hvdc_T

  • hvdc_Pf

  • bus_area_indices

Returns:

get_name_to_results_type_dict()[source]
Returns:

get_name_tree()[source]
Returns:

hvdc_F: ndarray[tuple[Any, ...], dtype[int64]] | None
hvdc_T: ndarray[tuple[Any, ...], dtype[int64]] | None
property is_3ph
mdl(result_type: ResultTypes) → ResultsTable[source]

Get results model (overloaded in the respective implementations) :param result_type: ResultTypes

name: str
original_sample_idx: ndarray[tuple[Any, ...], dtype[int64]] | None
parse_data(data: Dict[str, Any | Dict[str, Any]])[source]

The function to parse the data created with get_dict :param data: :return:

parse_saved_data(grid: ~VeraGridEngine.Devices.multi_circuit.MultiCircuit, data_dict: ~typing.Dict[str, ~pandas.DataFrame | ~numpy.ndarray], logger: ~VeraGridEngine.basic_structures.Logger = <VeraGridEngine.basic_structures.Logger object>) → None[source]
Parameters:

  • grid – MultiCircuit

  • data_dict – Dictionary with the info loaded from disk

  • logger – Logger

Returns:

plotting_allowed() → bool[source]
Returns:

prepare_for_saving() → None[source]

Refresh any derived payload that must be serialized with the results.

Returns:

None.

register(name: str, tpe: ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]] | ndarray[tuple[Any, ...], dtype[complex128]], old_names: None | List[str] = None) → None[source]

Runtime registration is intentionally disabled. :param name: name of the variable to register (is checked) :param tpe: type of the variable :param old_names: list of old names for retro compatibility (optional) :return: None

report_text: str
sampled_probabilities: ndarray[tuple[Any, ...], dtype[float64]] | None
study_results_type: StudyResultsType
property time_array: DatetimeIndex

Array of time steps :return:

time_indices: ndarray[tuple[Any, ...], dtype[int64]] | None
to_json(file_name)[source]

Export as json :param file_name: File name

using_clusters
class VeraGridEngine.Simulations.results_template.ResultsTemplateMeta(name: str, bases: Tuple[type, ...], namespace: Dict[str, Any])[source]

Bases: type

Metaclass that pre-builds inherited results schema declarations.

VeraGridEngine.Simulations.types module

Module contents