VeraGridEngine.Utils package

Subpackages

Submodules

VeraGridEngine.Utils.charatcter_detection module

VeraGridEngine.Utils.charatcter_detection.can_decode(data: bytes, encoding: str) → bool[source]

Check whether a byte sequence can be decoded with a specific encoding.

Parameters:

data – Raw byte sequence to validate.
encoding – Candidate text encoding.

Returns:

True when decoding succeeds, False otherwise.

VeraGridEngine.Utils.charatcter_detection.detect_character_encoding(source: str | Path | bytes, fallback_encodings: tuple[str, ...] | None = None) → str[source]

Detect a text encoding and validate the result by decoding the payload.

The algorithm starts with chardet because it is cheap and already available in the project. The algorithm then validates the detected encoding by actually decoding the full byte sequence. If the detected encoding fails, the algorithm tries a short sequence of explicit fallback encodings that commonly appear in imported text files.

Parameters:

source – File path or raw bytes to inspect.
fallback_encodings – Optional ordered list of fallback encodings.

Returns:

The first encoding that decodes the full payload.

VeraGridEngine.Utils.compare module

VeraGridEngine.Utils.compare.compare_arr(arr, arr_expected, tol: float, name: str, test: str, logger: Logger) → None[source]

VeraGridEngine.Utils.emt_boundary_update_wrapper module

class VeraGridEngine.Utils.emt_boundary_update_wrapper.BoundaryUpdateWrapper[source]

Bases: object

Lightweight interface for EMT boundary updates and forced events.

The class lives in VeraGridEngine.Utils so EMT templates, procedural logic and solver modules can import it statically without pulling one heavy solver or simulation package __init__ into the import graph.

get_next_forced_event_time(t_prev: float, t_target: float) → float | None[source]

Return the earliest forced event time in (t_prev, t_target].

Parameters:

t_prev – Previous solver time.
t_target – Nominal target time.

Returns:

Forced event time or None.

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Update the runtime parameter vector in place.

Parameters:

t – Current solver time.
x – Accepted state vector.
params – Runtime parameter vector.

Returns:

None.

VeraGridEngine.Utils.procedural_logic module

class VeraGridEngine.Utils.procedural_logic.AnalogFlipFlopLogic(output_var_name: str, input_expr: Expr | Comparison, set_expr: Expr | Comparison, reset_expr: Expr | Comparison, name: str = '')[source]

Bases: ProceduralLogicBase

Store an analog value when a set/reset latch enters the high state.

bind(problem: ProceduralProblem) → None[source]

Resolve the runtime output slot and reset the analog latch state.

Parameters:: problem – Bound EMT problem.
Returns:: None

held_value

initialized

input_expr

logic_tpe = 'analog_flipflop'

output_idx

output_var_name

remap(var_mapping: Dict[Expr | str, Expr]) → AnalogFlipFlopLogic[source]

Clone the logic under a variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped analog flip-flop logic.

reset_expr

set_expr

state

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Apply analog set/reset semantics and update the held value.

Parameters:

t – Current solver time.
x – Accepted state vector.
params – Runtime parameter vector.

Returns:

None

class VeraGridEngine.Utils.procedural_logic.BlockProceduralLogicUpdater(problem: ProceduralProblem, logic_entries: List[ProceduralLogicBase])[source]

Bases: BoundaryUpdateWrapper

Boundary updater that delegates runtime decisions to block-attached procedural logic entries.

get_next_forced_event_time(t_prev: float, t_target: float) → float | None[source]

Return the earliest forced event requested by any procedural logic entry.

Parameters:

t_prev – Previous solver time.
t_target – Nominal target time.

Returns:

Earliest event in the interval, if any.

initialize_modes(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Initialize retained procedural mode outputs at startup.

The generic EMT runtime-parameter initialization seeds mode_dict values before the procedural layer is active. For procedural outputs such as sampled values and saturations, the correct startup value is the value produced by the procedural logic itself when evaluated at the initialized operating point. This startup pass applies the same update logic once after the continuous initialization has populated the state, algebraic, and differential guesses, so retained mode outputs start from the exact operating-point sample that the runtime layer will hold.

Parameters:

t – Startup simulation time.
x – Initialized state/algebraic vector.
params – Runtime parameter vector updated in place.

Returns:

None

logic_entries

problem

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Forward one runtime update to all procedural logic entries.

Parameters:

t – Current solver time.
x – Accepted state vector.
params – Runtime parameter vector.

Returns:

None

class VeraGridEngine.Utils.procedural_logic.ConditionalDiagnosticLogic(condition_expr: Expr | Comparison, message: str, initialization_only: bool, name: str = '')[source]

Bases: ProceduralLogicBase

Preserve one conditional source-model diagnostic declaratively.

The entry deliberately owns no presentation side effect. Solver or UI code may later consume the typed condition and exact message according to its logging policy without making import-time data executable.

Parameters:

condition_expr – Boolean expression that activates the diagnostic.
message – Exact source diagnostic text.
initialization_only – Whether PowerFactory evaluates it only during initialization.
name – Optional diagnostic display name.

condition_expr: Expr | Comparison

initialization_only: bool

logic_tpe = 'conditional_diagnostic'

message: str

remap(var_mapping: Dict[Expr | str, Expr]) → ConditionalDiagnosticLogic[source]

Clone the diagnostic under a symbolic variable mapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped diagnostic declaration.

class VeraGridEngine.Utils.procedural_logic.DelayedSwitchEventLogic(output_var_name: str, guard_expr: Expr | Comparison, trigger_expr: Expr | Comparison, delay_expr: Expr | Comparison, target_device_idtag: str, target_switch_idtag: str, target_terminal_index: int, initial_closed: bool, command_closed: bool, name: str = '')[source]

Bases: ProceduralLogicBase

Apply one delayed switch command on a guarded rising trigger edge.

bind(problem: ProceduralProblem) → None[source]

Resolve the local switch-position mode and clear runtime state.

Parameters:: problem – Bound RMS or EMT problem.
Returns:: None.

command_closed: bool

delay_expr: Expr | Comparison

fired: bool

get_next_forced_event_time(t_prev: float, t_target: float) → float | None[source]

Return the exact delayed command boundary inside the next step.

Parameters:

t_prev – Previous solver time.
t_target – Nominal target time.

Returns:

Pending event time inside the interval, if any.

guard_expr: Expr | Comparison

initial_closed: bool

initialized: bool

logic_tpe = 'delayed_switch_event'

output_idx: int

output_var_name: str

pending_event_time: float | None

remap(var_mapping: Dict[Expr | str, Expr]) → DelayedSwitchEventLogic[source]

Clone the event under one symbolic variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped switch event.

state: float

target_device_idtag: str

target_switch_idtag: str

target_terminal_index: int

trigger_expr: Expr | Comparison

trigger_was_high: bool

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Detect a guarded rising edge and apply its delayed command once.

Parameters:

t – Current accepted solver time.
x – Current accepted state vector.
params – Runtime parameter vector updated in place.

Returns:

None.

class VeraGridEngine.Utils.procedural_logic.DelayedThresholdLatchLogic(monitored_var_name: str, mode_var_name: str, threshold: float, delay: float, reset_delay: float | None = None, name: str = '')[source]

Bases: ProceduralLogicBase

Comparator + timer + latch procedural logic for runtime modes.

The logic watches one state variable, arms a timer when a threshold is crossed, applies a mode change after a delay, and optionally resets after a second delay.

bind(problem: ProceduralProblem) → None[source]

Resolve the monitored state and runtime mode indices.

Parameters:: problem – Bound EMT problem.
Returns:: None

delay

get_next_forced_event_time(t_prev: float, t_target: float) → float | None[source]

Return the next exact trip or reset event inside one solver step.

Parameters:

t_prev – Previous solver time.
t_target – Nominal target time.

Returns:

First pending event in the interval, if any.

get_trace_arrays() → Tuple[ndarray, ndarray, ndarray, ndarray, ndarray, ndarray][source]

Return the internal trace arrays as dense numpy arrays.

Returns:: Tuple with times, measurements, comparator, timer, latch, and mode traces.

last_t_prev: float | None

logic_tpe = 'delayed_threshold_latch'

mode_idx

mode_var_name

monitored_idx

monitored_var_name

pending_reset_time: float | None

pending_trip_time: float | None

pickup_time: float | None

remap(var_mapping: Dict[Expr | str, Expr]) → DelayedThresholdLatchLogic[source]

Clone the logic under a variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped delayed-threshold latch logic.

reset_applied_solver_time: float | None

reset_applied_time: float | None

reset_delay

threshold

trace_comparator: List[float]

trace_latched: List[float]

trace_measure: List[float]

trace_mode: List[float]

trace_t: List[float]

trace_timer_armed: List[float]

trip_applied_solver_time: float | None

trip_applied_time: float | None

tripped

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Advance the delayed-threshold latch logic and tracing state.

Parameters:

t – Current solver time.
x – Accepted state vector.
params – Runtime parameter vector.

Returns:

None

class VeraGridEngine.Utils.procedural_logic.FixedSampleLogic(output_var_name: str, condition_expr: Expr | Comparison, name: str = '')[source]

Bases: ProceduralLogicBase

Retain the initial truth value of one condition in a runtime mode variable.

bind(problem: ProceduralProblem) → None[source]

Resolve the runtime output slot for this logic entry.

Parameters:: problem – Bound EMT problem.
Returns:: None

condition_expr

initialized

logic_tpe = 'fixed_sample'

output_idx

output_var_name

remap(var_mapping: Dict[Expr | str, Expr]) → FixedSampleLogic[source]

Clone the logic under a variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped fixed-sample logic.

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Sample the condition once and keep it fixed afterward.

Parameters:

t – Current solver time.
x – Accepted state vector.
params – Runtime parameter vector.

Returns:

None

class VeraGridEngine.Utils.procedural_logic.FlipFlopLogic(output_var_name: str, set_expr: Expr | Comparison, reset_expr: Expr | Comparison, name: str = '')[source]

Bases: ProceduralLogicBase

Store a binary set/reset latch in a runtime mode variable.

bind(problem: ProceduralProblem) → None[source]

Resolve the runtime output slot for the latch state.

Parameters:: problem – Bound EMT problem.
Returns:: None

initialized

logic_tpe = 'flipflop'

output_idx

output_var_name

remap(var_mapping: Dict[Expr | str, Expr]) → FlipFlopLogic[source]

Clone the logic under a variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped flip-flop logic.

reset_expr

set_expr

state

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Apply set/reset semantics and write the resulting latch state.

Parameters:

t – Current solver time.
x – Accepted state vector.
params – Runtime parameter vector.

Returns:

None

Bases: ProceduralLogicBase

Clamp the rate of change of one procedural value between lower and upper slopes.

bind(problem: ProceduralProblem) → None[source]

Bind the gradient-limited output and reset its runtime state.

Parameters:: problem – Runtime problem that owns variables and parameter indices.
Returns:: None.

held_value

initialized

last_time: float | None

logic_tpe = 'gradient_limiter'

lower_rate_expr

output_idx

output_var_name

remap(var_mapping: Dict[Expr | str, Expr]) → GradientLimiterLogic[source]

Clone one logic entry under a variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped procedural logic entry.

source_expr

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Update mode or event parameters before the Newton step.

Parameters:

t – Current solver time.
x – Current accepted state.
params – Runtime parameter vector to mutate in place.

Returns:

None

upper_rate_expr

class VeraGridEngine.Utils.procedural_logic.HardSaturationLogic(output_var_name: str, u_expr: Expr | Comparison, u_min_expr: Expr | Comparison, u_max_expr: Expr | Comparison, name: str = '')[source]

Bases: ProceduralLogicBase

Sample one input and apply hard saturation to a runtime mode variable.

bind(problem: ProceduralProblem) → None[source]

Bind the saturation output and numeric evaluators to a problem.

Parameters:: problem – Runtime problem that owns variables and parameter indices.
Returns:: None.

logic_tpe = 'hard_saturation'

output_idx

output_var_name

remap(var_mapping: Dict[Expr | str, Expr]) → HardSaturationLogic[source]

Clone one logic entry under a variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped procedural logic entry.

u_expr

u_max_expr

u_min_expr

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Update mode or event parameters before the Newton step.

Parameters:

t – Current solver time.
x – Current accepted state.
params – Runtime parameter vector to mutate in place.

Returns:

None

Bases: ProceduralLogicBase

Buffer-based moving average with optional delay before the averaging window.

bind(problem: ProceduralProblem) → None[source]

Bind the moving-average output and reset its sample history.

Parameters:: problem – Runtime problem that owns variables and parameter indices.
Returns:: None.

delay_expr

history: List[Tuple[float, float]]

logic_tpe = 'moving_average'

output_idx

output_var_name

remap(var_mapping: Dict[Expr | str, Expr]) → MovingAverageLogic[source]

Clone one logic entry under a variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped procedural logic entry.

source_expr

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Update mode or event parameters before the Newton step.

Parameters:

t – Current solver time.
x – Current accepted state.
params – Runtime parameter vector to mutate in place.

Returns:

None

window_expr

class VeraGridEngine.Utils.procedural_logic.PickupDropoffLogic(output_var_name: str, bool_expr: Expr | Comparison, pickup_delay_expr: Expr | Comparison, drop_delay_expr: Expr | Comparison, name: str = '', output_var_uid: int | None = None)[source]

Bases: ProceduralLogicBase

Implement a delayed pickup/dropoff relay with retained binary state.

bind(problem: ProceduralProblem) → None[source]

Resolve the runtime output slot and clear the relay timers.

Parameters:: problem – Bound EMT problem.
Returns:: None

bool_expr

drop_delay_expr

drop_started_at: float | None

get_next_forced_event_time(t_prev: float, t_target: float) → float | None[source]

Return the next exact pickup or dropoff event inside one step.

Parameters:

t_prev – Previous solver time.
t_target – Nominal target time.

Returns:

First pending relay event in the interval, if any.

initialized

logic_tpe = 'pickup_dropoff'

output_idx

output_var_name

output_var_uid: int | None

pending_drop_time: float | None

pending_pickup_time: float | None

pickup_delay_expr

pickup_started_at: float | None

remap(var_mapping: Dict[Expr | str, Expr]) → PickupDropoffLogic[source]

Clone the logic under a variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped pickup/dropoff logic.

state

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Advance the relay timers and binary state.

Parameters:

t – Current solver time.
x – Accepted state vector.
params – Runtime parameter vector.

Returns:

None

class VeraGridEngine.Utils.procedural_logic.ProceduralLogicBase(name: str = '')[source]

Bases: ProceduralLogicEntryContract[Expr]

Base class for procedural logic objects attached to symbolic blocks.

Procedural logic is evaluated outside the compiled residual kernels and is intended to drive runtime modes, retained flags, timers, or event scheduling in a structured way.

bind(problem: ProceduralProblem) → None[source]

Bind the logic to a concrete EMT problem.

Parameters:: problem – EMT problem owning the block and variable maps.
Returns:: None

evaluate_numeric_expression(expr: Expr | Comparison | float | int | bool, t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → float[source]

Evaluate a procedural expression against the accepted EMT state.

Parameters:

expr – Expression or comparison to evaluate.
t – Current solver time.
x – Accepted state vector.
params – Runtime parameter vector.

Returns:

Numeric value of the expression.

get_next_forced_event_time(t_prev: float, t_target: float) → float | None[source]

Return the next exact event time inside the interval if known.

Parameters:

t_prev – Previous solver time.
t_target – Nominal target time.

Returns:

Event time or None.

logic_tpe = 'base'

name

remap(var_mapping: Dict[Expr | str, Expr]) → ProceduralLogicBase[source]

Clone one logic entry under a variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped procedural logic entry.

to_data() → Dict[str, object][source]

Return the version-independent declarative logic representation.

Returns:: Typed declarative logic data.

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Update mode or event parameters before the Newton step.

Parameters:

t – Current solver time.
x – Current accepted state.
params – Runtime parameter vector to mutate in place.

Returns:

None

class VeraGridEngine.Utils.procedural_logic.ProceduralLogicCodec(*args, **kwargs)[source]

Bases: ProceduralLogicCodecContract[Expr]

Reconstruct procedural state machines at an explicit parse boundary.

parse_entries(entries: Iterable[Dict[str, object]]) → Iterable[ProceduralLogicEntryContract[Expr]][source]

Build typed state machines from declarative entry data.

Parameters:: entries – Declarative procedural entries.
Returns:: Reconstructed procedural state machines.

class VeraGridEngine.Utils.procedural_logic.ProceduralNumericEvaluator(logic: ProceduralLogicBase, expression: Expr | Comparison | float | int | bool)[source]

Bases: object

Evaluate one procedural expression against an accepted solver snapshot.

Parameters:

logic – Bound procedural state machine that owns sample-time rules.
expression – Declarative expression evaluated by the wrapper.

class VeraGridEngine.Utils.procedural_logic.ProceduralProblem(*args, **kwargs)[source]

Bases: Protocol

Declare the EMT surface required by procedural runtime logic.

property event_params_values: ndarray[tuple[Any, ...], dtype[float64]]

Return the current mutable runtime-parameter vector.

Returns:: Runtime values updated by procedural boundary logic.

get_parameters_values() → List[Const][source]: Return constant parameter values in compiled order.

get_var_idx(variable: Var) → int[source]

Return the compiled index for one symbolic variable.

Parameters:: variable – Symbolic variable whose solver-vector index is needed.
Returns:: Zero-based index in the compiled solver vector.

get_variable_parameter_number() → int[source]: Return the number of mutable runtime parameters.

get_x0() → ndarray[tuple[Any, ...], dtype[float64]][source]: Return the current initialization vector.

property glob_time: Var

Return the symbolic variable representing global solver time.

Returns:: Canonical symbolic solver-time variable.

property sys_block: Block

Return the canonical root block compiled by the runtime problem.

Returns:: Root block owning all attached procedural declarations.

property uid2idx_event_params: Dict[int, int]

Return mutable runtime-parameter indexes keyed by symbolic UID.

Returns:: Runtime-parameter indexes keyed by stable symbolic UID.

property uid2idx_params: Dict[int, int]

Return constant-parameter indexes keyed by symbolic UID.

Returns:: Constant-parameter indexes keyed by stable symbolic UID.

property uid2idx_vars: Dict[int, int]

Return compiled state-variable indexes keyed by symbolic UID.

Returns:: State-vector indexes keyed by stable symbolic UID.

class VeraGridEngine.Utils.procedural_logic.ResetOnRisingEdgeLogic(target_var_name: str, reset_expr: Expr | Comparison, value_expr: Expr | Comparison, name: str = '')[source]

Bases: ProceduralLogicBase

Apply one value reset on the rising edge of a procedural condition.

bind(problem: ProceduralProblem) → None[source]

Resolve whether the reset target is a state variable or a runtime parameter.

Parameters:: problem – Bound EMT problem.
Returns:: None

initialized

last_reset_high

logic_tpe = 'reset_on_rising_edge'

remap(var_mapping: Dict[Expr | str, Expr]) → ResetOnRisingEdgeLogic[source]

Clone the logic under a variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped reset logic.

reset_expr

target_param_idx

target_state_idx

target_var_name

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Apply the reset value only on the rising edge of the reset condition.

Parameters:

t – Current solver time.
x – Accepted state vector.
params – Runtime parameter vector.

Returns:

None

value_expr

class VeraGridEngine.Utils.procedural_logic.SampledValueLogic(output_var_name: str, source_expr: Expr | Comparison, name: str = '', output_var_uid: int | None = None)[source]

Bases: ProceduralLogicBase

Sample one expression at each accepted step and store it in a runtime mode variable.

bind(problem: ProceduralProblem) → None[source]

Resolve the runtime output slot for this sampled value.

Parameters:: problem – Bound EMT problem.
Returns:: None

logic_tpe = 'sampled_value'

output_idx: int

output_var_name: str

output_var_uid: int | None

remap(var_mapping: Dict[Expr | str, Expr]) → SampledValueLogic[source]

Clone the logic under a variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped sampled-value logic.

source_expr: Expr | Comparison

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Refresh the sampled value using the accepted state.

Parameters:

t – Current solver time.
x – Accepted state vector.
params – Runtime parameter vector.

Returns:

None

class VeraGridEngine.Utils.procedural_logic.StartupHandoverLogic(mode_var_name: str, enable_time_var_name: str, on_value_var_name: str | None = None, name: str = '')[source]

Bases: ProceduralLogicBase

One-shot startup handover for hybrid EMT converter models.

The logic keeps one retained runtime mode at 0 during the averaged-startup interval and flips it to 1 exactly at t_enable. This lets the DAE stay continuous before the handover while still forcing an exact solver split when the switched bridge becomes electrically active.

bind(problem: ProceduralProblem) → None[source]

Resolve the retained mode and enable-time runtime indices.

Parameters:: problem – Bound EMT problem.
Returns:: None.

enable_time_idx

enable_time_var_name

get_next_forced_event_time(t_prev: float, t_target: float) → float | None[source]

Return the exact startup-handover time when it falls inside one solver step.

Parameters:

t_prev – Previous solver time.
t_target – Nominal target time.

Returns:

Exact handover time or None.

logic_tpe = 'startup_handover'

mode_idx

mode_var_name

on_value_idx

on_value_var_name

remap(var_mapping: Dict[Expr | str, Expr]) → StartupHandoverLogic[source]

Clone the logic under a variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped startup-handover logic.

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Update the retained startup mode from the current solver time.

Parameters:

t – Current solver time.
x – Accepted state vector.
params – Runtime parameter vector.

Returns:

None.

class VeraGridEngine.Utils.procedural_logic.ThreePhaseCarrierPwmLogic(mod_a_var_name: str, mod_b_var_name: str, mod_c_var_name: str, gate_a_mode_var_name: str, gate_b_mode_var_name: str, gate_c_mode_var_name: str, omega_sw_var_name: str, carrier_phase_var_name: str, name: str = '')[source]

Bases: ProceduralLogicBase

Regularly sampled three-phase carrier PWM logic.

The logic samples the modulation references at each carrier half-period, computes the exact gate transitions inside that interval, and stores the gate values as retained runtime modes. The symbolic DAE therefore sees piecewise constant gate parameters instead of symbolic comparator expressions.

bind(problem: ProceduralProblem) → None[source]

Resolve the variable indices and initialize the first PWM interval.

Parameters:: problem – Bound EMT problem.
Returns:: None.

carrier_phase_idx

carrier_phase_var_name

current_gate

gate_a_idx

gate_a_mode_var_name

gate_b_idx

gate_b_mode_var_name

gate_c_idx

gate_c_mode_var_name

get_next_forced_event_time(t_prev: float, t_target: float) → float | None[source]

Return the next PWM gate transition or carrier-boundary event.

Parameters:

t_prev – Previous solver time.
t_target – Nominal target time.

Returns:

Earliest event in the interval, if any.

initialized

interval_end_time: float | None

last_omega_sw: float | None

logic_tpe = 'three_phase_carrier_pwm'

mod_a_idx

mod_a_var_name

mod_b_idx

mod_b_var_name

mod_c_idx

mod_c_var_name

omega_sw_idx

omega_sw_var_name

pending_transition_gate

pending_transition_time

remap(var_mapping: Dict[Expr | str, Expr]) → ThreePhaseCarrierPwmLogic[source]

Clone the PWM logic under a variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped PWM logic.

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Advance the PWM gate states from the accepted modulation references.

Parameters:

t – Current solver time.
x – Accepted EMT state vector.
params – Runtime parameter vector.

Returns:

None.

class VeraGridEngine.Utils.procedural_logic.ThreePhaseCarrierSampledModulationLogic(mod_a_var_name: str, mod_b_var_name: str, mod_c_var_name: str, sample_a_mode_var_name: str, sample_b_mode_var_name: str, sample_c_mode_var_name: str, omega_sw_var_name: str, carrier_phase_var_name: str, name: str = '')[source]

Bases: ProceduralLogicBase

Regular-sampled three-phase modulation holder using carrier half-period boundaries.

The sampling instants intentionally follow ThreePhaseCarrierPwmLogic so a symbolic soft comparator can consume the same held m_a/m_b/m_c values as the procedural PWM scheduler.

bind(problem: ProceduralProblem) → None[source]

Resolve modulation, sampled-mode and carrier parameter indices.

Parameters:: problem – Concrete EMT problem.
Returns:: None.

carrier_phase_idx

carrier_phase_var_name

get_next_forced_event_time(t_prev: float, t_target: float) → float | None[source]

Force a solver event at each carrier half-period boundary.

Parameters:

t_prev – Previous accepted solver time.
t_target – Nominal target time.

Returns:

Next half-period boundary if it lies inside the step.

initialized

interval_end_time

logic_tpe = 'three_phase_carrier_sampled_modulation'

mod_a_idx

mod_a_var_name

mod_b_idx

mod_b_var_name

mod_c_idx

mod_c_var_name

omega_sw_idx

omega_sw_var_name

remap(var_mapping: Dict[Expr | str, Expr]) → ThreePhaseCarrierSampledModulationLogic[source]

Clone this logic under a variable-name remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped sampled-modulation logic.

sample_a_idx

sample_a_mode_var_name

sample_b_idx

sample_b_mode_var_name

sample_c_idx

sample_c_mode_var_name

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Sample modulation at initialization and at carrier half-period boundaries.

Parameters:

t – Current solver time.
x – Accepted EMT state vector.
params – Runtime parameter vector to mutate.

Returns:

None.

Bases: ProceduralLogicBase

Sample one input and expose its delayed value through a runtime mode variable.

bind(problem: ProceduralProblem) → None[source]

Bind the delayed output and reset its sample history.

Parameters:: problem – Runtime problem that owns variables and parameter indices.
Returns:: None.

delay_expr

history: List[Tuple[float, float]]

logic_tpe = 'time_delay'

output_idx

output_var_name

remap(var_mapping: Dict[Expr | str, Expr]) → TimeDelayLogic[source]

Clone one logic entry under a variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped procedural logic entry.

source_expr

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Update mode or event parameters before the Newton step.

Parameters:

t – Current solver time.
x – Current accepted state.
params – Runtime parameter vector to mutate in place.

Returns:

None

class VeraGridEngine.Utils.procedural_logic.ValveStateLogic(mode_var_name: str, valve_type_var_name: str, gate_var_name: str, antiparallel_var_name: str, voltage_eps_var_name: str, current_eps_var_name: str, valve_voltage_var_name: str, valve_current_var_name: str, name: str = '')[source]

Bases: ProceduralLogicBase

Retained conduction-state logic for generic EMT valve blocks.

The logic updates a single retained runtime mode that encodes the active path:

+1: forward conduction
0: blocked state
-1: reverse conduction through the antiparallel path

antiparallel_idx

antiparallel_var_name

bind(problem: ProceduralProblem) → None[source]

Resolve the runtime and algebraic indices required by the valve logic.

Parameters:: problem – Bound EMT problem.
Returns:: None.

current_eps_idx

current_eps_var_name

gate_idx

gate_var_name

logic_tpe = 'valve_state'

mode_idx

mode_var_name

remap(var_mapping: Dict[Expr | str, Expr]) → ValveStateLogic[source]

Clone the valve logic under a variable remapping.

Parameters:: var_mapping – Variable substitution map.
Returns:: Remapped valve logic.

update(t: float, x: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]]) → None[source]

Update the retained conduction-path mode from the accepted EMT state.

Parameters:

t – Current solver time.
x – Accepted EMT state vector.
params – Flat runtime parameter vector.

Returns:

None.

valve_current_idx

valve_current_var_name

valve_type_idx

valve_type_var_name

valve_voltage_idx

valve_voltage_var_name

voltage_eps_idx

voltage_eps_var_name

VeraGridEngine.Utils.procedural_logic.aflipflop(x: Expr | Comparison, boolset: Expr | Comparison, boolreset: Expr | Comparison, output: Var | str, name: str = '') → AnalogFlipFlopLogic[source]

Build the procedural equivalent of PowerFactory aflipflop.

Parameters:

x – Analog expression to store while the internal state is high.
boolset – Set condition.
boolreset – Reset condition.
output – Runtime mode variable receiving the held analog value.
name – Optional logic name.

Returns:

Analog flip-flop procedural logic entry.

VeraGridEngine.Utils.procedural_logic.bool_and(*args: Expr | Comparison | float | int) → Expr[source]

Build the boolean AND of one or more procedural expressions.

Parameters:: args – Boolean-like expressions encoded as 0/1 values.
Returns:: Expression equal to 1.0 only when all inputs are true.

Build the exclusive-OR of two procedural expressions.

Parameters:

left – Left boolean-like expression.
right – Right boolean-like expression.

Returns:

Expression equal to 1.0 only when exactly one input is true.

VeraGridEngine.Utils.procedural_logic.bool_nand(*args: Expr | Comparison | float | int) → Expr[source]

Build the boolean NAND of one or more procedural expressions.

Parameters:: args – Boolean-like expressions encoded as 0/1 values.
Returns:: Expression equal to the negated AND of the inputs.

VeraGridEngine.Utils.procedural_logic.bool_nor(*args: Expr | Comparison | float | int) → Expr[source]

Build the boolean NOR of one or more procedural expressions.

Parameters:: args – Boolean-like expressions encoded as 0/1 values.
Returns:: Expression equal to the negated OR of the inputs.

VeraGridEngine.Utils.procedural_logic.bool_not(arg: Expr | Comparison | float | int) → Expr[source]

Build the boolean NOT of one procedural expression.

Parameters:: arg – Boolean-like expression encoded as 0/1.
Returns:: Expression equal to 1.0 when the input is false.

VeraGridEngine.Utils.procedural_logic.bool_or(*args: Expr | Comparison | float | int) → Expr[source]

Build the boolean OR of one or more procedural expressions.

Parameters:: args – Boolean-like expressions encoded as 0/1 values.
Returns:: Expression equal to 1.0 when at least one input is true.

VeraGridEngine.Utils.procedural_logic.build_boundary_updater_from_block(problem: ProceduralProblem) → BlockProceduralLogicUpdater | None[source]

Build an isolated runtime updater from the logic attached to problem.sys_block.

The block hierarchy is canonical persistent model data. Runtime binding must therefore target reconstructed entries so solver-owned objects, such as progress signals, cannot become reachable from the persistent model.

Parameters:: problem – EMT problem containing the root block.
Returns:: Boundary updater or None.

VeraGridEngine.Utils.procedural_logic.build_procedural_logic_entry(data: Dict[str, object]) → ProceduralLogicBase[source]

Deserialize one procedural logic entry.

Parameters:: data – Serialized logic config.
Returns:: Procedural logic object.

VeraGridEngine.Utils.procedural_logic.clone_procedural_logic_entries(entries: Iterable[ProceduralLogicEntryContract[Expr]], var_mapping: Dict[Expr | str, Expr]) → List[ProceduralLogicEntryContract[Expr]][source]

Clone procedural logic entries under a variable remapping.

Parameters:

entries – Source procedural logic entries.
var_mapping – Mapping from old variables/names to remapped expressions.

Returns:

Remapped procedural logic entries.

Build the procedural equivalent of PowerFactory delay.

Parameters:

input_expr – Source expression to be delayed.
T – Delay in seconds.
output – Runtime mode variable receiving the delayed value.
name – Optional logic name.

Returns:

Time-delay procedural logic entry.

VeraGridEngine.Utils.procedural_logic.fixed_sample(output: Var | str, when: Expr | Comparison, name: str = '') → FixedSampleLogic[source]

Build a fixed-sample selector that stores a boolean condition at initialization.

Parameters:

output – Runtime mode variable receiving the retained value.
when – Condition sampled at initialization time.
name – Optional logic name.

Returns:

Fixed-sample procedural logic entry.

VeraGridEngine.Utils.procedural_logic.flipflop(boolset: Expr | Comparison, boolreset: Expr | Comparison, output: Var | str, name: str = '') → FlipFlopLogic[source]

Build the procedural equivalent of PowerFactory flipflop.

Parameters:

boolset – Set condition.
boolreset – Reset condition.
output – Runtime mode variable storing the logical state.
name – Optional logic name.

Returns:

Flip-flop procedural logic entry.

Build the procedural equivalent of PowerFactory gradlim_const.

Parameters:

input_expr – Source expression to be rate-limited.
gradmin – Lower slope limit.
gradmax – Upper slope limit.
output – Runtime mode variable receiving the limited value.
name – Optional logic name.

Returns:

Gradient-limiter procedural logic entry.

VeraGridEngine.Utils.procedural_logic.hard_saturation(output: Var | str, u: Expr | Comparison, u_min: Expr | Comparison, u_max: Expr | Comparison, name: str = '') → HardSaturationLogic[source]

Build one procedural hard-saturation entry updated outside Newton residuals.

Parameters:

output – Runtime mode variable receiving saturated value.
u – Unsaturated input expression.
u_min – Lower saturation bound.
u_max – Upper saturation bound.
name – Optional logic name.

Returns:

Hard-saturation procedural logic entry.

Reject equation-level ifelse usage in the public procedural API.

Parameters:

boolexpr – Switching condition.
when_true – Value chosen for the true branch.
when_false – Value chosen for the false branch.

Returns:

Never returns because public equation-level selection is blocked.

VeraGridEngine.Utils.procedural_logic.lastvalue(input_expr: Expr | Comparison, output: Var | str, name: str = '') → SampledValueLogic[source]

Build the procedural equivalent of PowerFactory lastvalue.

Parameters:

input_expr – Source expression to be sampled.
output – Runtime mode variable receiving the sampled value.
name – Optional logic name.

Returns:

Sampled-value procedural logic entry.

Build the procedural equivalent of PowerFactory movingavg.

Parameters:

input_expr – Source expression to be averaged.
Tdel – Delay before the averaging window.
Tlength – Averaging window length in seconds.
output – Runtime mode variable receiving the averaged value.
name – Optional logic name.

Returns:

Moving-average procedural logic entry.

VeraGridEngine.Utils.procedural_logic.picdro(boolexpr: Expr | Comparison, Tpick: Expr | Comparison, Tdrop: Expr | Comparison, output: Var | str, name: str = '') → PickupDropoffLogic[source]

Build the procedural equivalent of PowerFactory picdro.

Parameters:

boolexpr – Relay driving condition.
Tpick – Pickup delay.
Tdrop – Dropoff delay.
output – Runtime mode variable storing the relay state.
name – Optional logic name.

Returns:

Pickup/dropoff procedural logic entry.

VeraGridEngine.Utils.procedural_logic.picdro_const(boolexpr: Expr | Comparison, Tpick: Expr | Comparison, Tdrop: Expr | Comparison, output: Var | str, name: str = '') → PickupDropoffLogic[source]

Build the procedural equivalent of PowerFactory picdro_const.

Parameters:

boolexpr – Relay driving condition.
Tpick – Pickup delay.
Tdrop – Dropoff delay.
output – Runtime mode variable storing the relay state.
name – Optional logic name.

Returns:

Pickup/dropoff procedural logic entry.

VeraGridEngine.Utils.procedural_logic.pickup_dropoff(output: Var | str, boolexpr: Expr | Comparison, Tpick: Expr | Comparison, Tdrop: Expr | Comparison, name: str = '') → PickupDropoffLogic[source]

Build a pickup/dropoff relay-style procedural logic entry.

Parameters:

output – Runtime mode variable storing the relay state.
boolexpr – Pickup/reset driving condition.
Tpick – Pickup delay.
Tdrop – Dropoff delay.
name – Optional logic name.

Returns:

Pickup/dropoff procedural logic entry.

VeraGridEngine.Utils.procedural_logic.procedural_logic_entry_to_dict(entry: ProceduralLogicBase) → Dict[str, object][source]

Serialize one procedural logic entry.

Parameters:: entry – Procedural logic entry.
Returns:: Serialized logic dictionary.

VeraGridEngine.Utils.procedural_logic.procedural_logic_from_dict(entries: List[Dict[str, object] | ProceduralLogicBase]) → List[ProceduralLogicBase][source]

Deserialize a list of procedural logic entries.

Parameters:: entries – Serialized entries.
Returns:: Procedural logic objects.

VeraGridEngine.Utils.procedural_logic.procedural_logic_to_dict(entries: List[ProceduralLogicBase]) → List[Dict[str, object]][source]

Serialize a list of procedural logic entries.

Parameters:: entries – Logic entries.
Returns:: Serialized representation.

VeraGridEngine.Utils.procedural_logic.reset(var: Var | str, rst: Expr | Comparison, val: Expr | Comparison, name: str = '') → ResetOnRisingEdgeLogic[source]

Build the procedural equivalent of PowerFactory reset.

Parameters:

var – Target runtime/state variable name.
rst – Reset trigger condition.
val – Value applied on the rising edge.
name – Optional logic name.

Returns:

Reset-on-rising-edge procedural logic entry.

VeraGridEngine.Utils.procedural_logic.sampled_value(output: Var | str, source: Expr | Comparison, name: str = '') → SampledValueLogic[source]

Build a sampled runtime value updated outside the Newton residual.

Parameters:

output – Runtime mode variable receiving the sampled value.
source – Source expression to be evaluated at the accepted sample time.
name – Optional logic name.

Returns:

Sampled-value procedural logic entry.

Reject equation-level select usage in the public procedural API.

Parameters:

boolexpr – Switching condition.
when_true – Value chosen for the true branch.
when_false – Value chosen for the false branch.

Returns:

Never returns because public equation-level selection is blocked.

VeraGridEngine.Utils.procedural_logic.selfix(boolexpr: Expr | Comparison, output: Var | str, name: str = '') → FixedSampleLogic[source]

Build the procedural equivalent of PowerFactory selfix.

Parameters:

boolexpr – Condition sampled at initialization time.
output – Runtime mode variable receiving the retained value.
name – Optional logic name.

Returns:

Fixed-sample procedural logic entry.

VeraGridEngine.Utils.procedural_logic.selfix_const(boolexpr: Expr | Comparison, output: Var | str, name: str = '') → FixedSampleLogic[source]

Build the procedural equivalent of PowerFactory selfix_const.

Parameters:

boolexpr – Condition sampled at initialization time.
output – Runtime mode variable receiving the retained value.
name – Optional logic name.

Returns:

Fixed-sample procedural logic entry.

VeraGridEngine.Utils.procedural_logic.startup_handover(mode: Var | str, t_enable: Var | str, name: str = '', on_value: Var | str | None = None) → StartupHandoverLogic[source]

Build one exact-time startup-handover logic entry.

Parameters:

mode – Retained mode variable that becomes 1 after the handover time.
t_enable – Runtime parameter storing the switching-enable time.
name – Optional logic name.
on_value – Optional runtime parameter whose value is copied after handover.

Returns:

Startup-handover procedural logic entry.

VeraGridEngine.Utils.progress_bar module

VeraGridEngine.Utils.progress_bar.print_progress_bar(iteration: int, total: int, length=40, txt='')[source]: Simple text progress bar :param iteration: current iteration (1 based) :param total: total progress :param length: length of the bar in characters :param txt: text to print at the end of the progress bar

VeraGridEngine.Utils.rms_models_types module

VeraGridEngine.Utils.rms_models_types.build_equivalence_classes_dict(grid: MultiCircuit) → Tuple[Dict[int, List[int]], Dict[int, List[List[int]]], Dict[int, List[int]], Dict[int, List[int]]][source]

this functions receives the grid and return a dictionary with clases of equivalence of the dynamic models and a dictionary with clases of equivalence of their variables. :param grid: :type grid: :return: Tuple of:

Dict with new uid as keys and lists of equivalent block uids as values

Dict with representative uid as keys and lists of lists of equivalent variable uids as values

Dict with representative uid as keys and lists of all block uids composing that model (including children)

Dict with representative uid as keys and lists of all block uids in the reference class (equivalent blocks and their children recursively, plus the representative and its children)

Return type:

VeraGridEngine.Utils.scores module

class VeraGridEngine.Utils.scores.TechnoEconomicScores[source]

Bases: object

InvestmentScores

arr() → ndarray[tuple[Any, ...], dtype[float64]][source]

Return multidimensional metrics for the optimization.

CAPEX and OPEX are reported as two separate entries, matching the objective formulation documented in investment_optimization.md, so that the search sees the investment cost and the operational cost as two independent magnitudes.

Returns:: array of 6 values: losses, overload, voltage module, voltage angle, CAPEX and OPEX

capex_score: float

property financial_score: float: Get the financial score: CAPEX + OPEX :return: float

losses_score: float

opex_score: float

overload_score: float

property tech_score

voltage_angle_score: float

voltage_module_score: float

VeraGridEngine.Utils.scores.get_overload_score(loading: ndarray[tuple[Any, ...], dtype[complex128]], branches_cost: ndarray[tuple[Any, ...], dtype[float64]], threshold=1.0) → float[source]: Compute overload score by multiplying the loadings above 100% by the associated branch cost. :param loading: load results :param branches_cost: all branch elements from studied grid :param threshold: threshold for overload :return: sum of all costs associated to branch overloads

VeraGridEngine.Utils.scores.get_voltage_module_score(voltage: ndarray[tuple[Any, ...], dtype[complex128]], vm_cost: ndarray[tuple[Any, ...], dtype[float64]], vm_max: ndarray[tuple[Any, ...], dtype[float64]], vm_min: ndarray[tuple[Any, ...], dtype[float64]]) → float[source]: Compute voltage module score by multiplying the voltages outside limits by the associated bus costs. :param voltage: voltage results :param vm_cost: Vm cost array :param vm_max: maximum voltage :param vm_min: minimum voltage :return: sum of all costs associated to voltage module deviation

VeraGridEngine.Utils.scores.get_voltage_phase_score(voltage: ndarray[tuple[Any, ...], dtype[complex128]], va_cost: ndarray[tuple[Any, ...], dtype[float64]], va_max: ndarray[tuple[Any, ...], dtype[float64]], va_min: ndarray[tuple[Any, ...], dtype[float64]]) → float[source]: Compute voltage phase score by multiplying the phases outside limits by the associated bus costs. :param voltage: voltage results :param va_cost: array of bus angles costs :param va_max: maximum voltage angles :param va_min: minimum voltage angles :return: sum of all costs associated to voltage module deviation

VeraGridEngine.Utils.zip_file_mgmt module

VeraGridEngine.Utils.zip_file_mgmt.open_data_frame_from_zip(file_name_zip, file_name)[source]: Open the csv files from a zip file :param file_name_zip: name of the zip file :param file_name: name of the file within the zip file :return: DataFrame

VeraGridEngine.Utils package

Subpackages

Submodules

VeraGridEngine.Utils.charatcter_detection module

VeraGridEngine.Utils.compare module

VeraGridEngine.Utils.emt_boundary_update_wrapper module

VeraGridEngine.Utils.procedural_logic module

VeraGridEngine.Utils.progress_bar module

VeraGridEngine.Utils.rms_models_types module

VeraGridEngine.Utils.scores module

VeraGridEngine.Utils.zip_file_mgmt module

Module contents