VeraGridEngine.Simulations.EMT.solvers package

Submodules

VeraGridEngine.Simulations.EMT.solvers.StructuralVectorizedSolver module

class VeraGridEngine.Simulations.EMT.solvers.StructuralVectorizedSolver.DirectResidualDispatcher(kernel: Callable[[...], Any])

Bases: object

Residual dispatcher backed by one monolithic in-place kernel.

evaluate(states: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]], history: ndarray[tuple[Any, ...], dtype[float64]], d_history: ndarray[tuple[Any, ...], dtype[float64]], h: float, history2: ndarray[tuple[Any, ...], dtype[float64]], out: ndarray[tuple[Any, ...], dtype[float64]]) → None

Evaluate the full residual system in place.

Parameters:

• states (Vec) – Current Newton iterate.

• params (Vec) – Full parameter vector.

• history (Vec) – Previous accepted state.

• d_history (Vec) – Previous derivative vector.

• h (float) – Effective local time step.

• history2 (Vec) – Secondary state history.

• out (Vec) – Residual output buffer.

Returns:

None.

Return type:

None

class VeraGridEngine.Simulations.EMT.solvers.StructuralVectorizedSolver.EquationGroup

Bases: object

Data structure to hold clustered equations for auto-vectorization. Replaces the generic dictionary for strict type hinting and safety.

add_indices(var_indices: List[int], row_index: int) → None

Adds matrix indices and row index to the cluster.

get_idx_matrix() → List[List[int]]

Returns the index matrix.

get_row_indices() → List[int]

Returns the row indices.

get_template_eq() → Expr | None

Returns the template equation.

get_template_vars() → List[Var] | None

Returns the template variables.

set_template(template_eq: Expr, template_vars: List[Var]) → None

Sets the structural template for this group.

class VeraGridEngine.Simulations.EMT.solvers.StructuralVectorizedSolver.FusedResidualDispatcher(kernel_specs: List[VectorizedKernelSpec])

Bases: object

Dispatcher that applies all vectorized residual kernels in order.

The structural group kernels are already eagerly compiled with Numba. The dispatcher keeps one reusable work row per group so the hot loop does not allocate temporary residual blocks.

evaluate(states: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]], history: ndarray[tuple[Any, ...], dtype[float64]], d_history: ndarray[tuple[Any, ...], dtype[float64]], h: float, history2: ndarray[tuple[Any, ...], dtype[float64]], out: ndarray[tuple[Any, ...], dtype[float64]]) → None

Evaluate the grouped residual system in place.

Parameters:

• states (Vec) – Current Newton iterate.

• params (Vec) – Full parameter vector.

• history (Vec) – Previous accepted state.

• d_history (Vec) – Previous derivative vector.

• h (float) – Effective local time step.

• history2 (Vec) – Secondary state history.

• out (Vec) – Residual output buffer.

Returns:

None.

Return type:

None

class VeraGridEngine.Simulations.EMT.solvers.StructuralVectorizedSolver.LegacyResidualEvaluatorAdapter(residual_callable: Callable[[...], Any], vec_flat_args: ndarray[tuple[Any, ...], dtype[float64]])

Bases: object

Adapt one legacy 8-argument residual callable to the fixed 7-argument solver interface.

class VeraGridEngine.Simulations.EMT.solvers.StructuralVectorizedSolver.Predictor(n_states: int)

Bases: object

Computes an explicit predictor for the Newton initial guess x_iter.

n_states

predict(x_iter: ndarray[tuple[Any, ...], dtype[float64]], x_prev: ndarray[tuple[Any, ...], dtype[float64]], dx_prev: ndarray[tuple[Any, ...], dtype[float64]], h: float, pred_method: DynamicIntegrationMethod | None = DAE_BackEuler) → ndarray[tuple[Any, ...], dtype[float64]]

Apply predictor in-place and return x_iter. :param x_iter: Array to be written with the predictor guess (in-place). :param x_prev: Previous full variable vector at time n. :param dx_prev: Previous derivative vector (only first n_states are meaningful). :param h: Time step. :param pred_method: Integration method enum/value at the prediction step :return:

class VeraGridEngine.Simulations.EMT.solvers.StructuralVectorizedSolver.SparseADJacobian(equations: ~typing.List[~VeraGridEngine.Utils.Symbolic.symbolic.Expr], variables: ~typing.List[~VeraGridEngine.Utils.Symbolic.symbolic.Var], parameters: ~typing.List[~VeraGridEngine.Utils.Symbolic.symbolic.Var], method: ~VeraGridEngine.enumerations.DynamicIntegrationMethod, use_cse: bool = True, dtype: ~typing.Any = <class 'numpy.float64'>, verbose: bool = False)

Bases: object

J

col_ptr

col_rows

color_cols

color_groups

color_ptr

colors

data_idx

dtype

equations

get_data_buffer() → ndarray[tuple[Any, ...], dtype[float64]]

Return the reusable CSC numeric buffer.

Returns:

CSC numeric buffer.

Return type:

Vec

get_matrix() → csc_matrix

Return the reusable CSC Jacobian shell.

Returns:

CSC Jacobian shell.

Return type:

csc_matrix

method

n_colors

n_cols

n_rows

parameters

row_idx

var_map: Dict[int, int]

variables

verbose

class VeraGridEngine.Simulations.EMT.solvers.StructuralVectorizedSolver.StructuralVectorizedMasterJacobianDispatcher(ad_kernel: Callable[[...], Any], n_colors: int)

Bases: object

Callable dispatcher that evaluates one generic AD kernel across all colors.

class VeraGridEngine.Simulations.EMT.solvers.StructuralVectorizedSolver.StructuralVectorizedSolver(problem: EmtProblemTemplate, t0: float, t_end: float, h: float, method: DynamicIntegrationMethod = DAE_Trapezoidal, pred_method: DynamicIntegrationMethod = None, dense_threshold: int = 100, verbose: bool = False, newton_max_iter: int = 20, auto_vectorization: bool = True, sparse_solver_backend_provider: SparseLinearSolverBackendProvider | None = None, newton_diag_config: NewtonDiagnosticsConfig | None = None)

Bases: object

auto_detect_vectorization(method: DynamicIntegrationMethod | None = None) → None

Infers the algebraic structure of the DAE system (clustering) and compiles the vectorized matrix kernels.

dense_threshold

fused_residual

get_backend_build_stats() → Dict[str, float]

Return setup timings collected during backend compilation.

Returns:

Backend build statistics.

Return type:

Dict[str, float]

get_last_runtime_stats() → Dict[str, float]

Return runtime statistics collected during the latest simulation.

Returns:

Runtime statistics.

Return type:

Dict[str, float]

get_last_sim_loop_time() → float

Return the latest integration-loop wall time.

Returns:

Latest loop time in seconds.

Return type:

float

h

jit_jacobians_ad: Dict[DynamicIntegrationMethod, SparseADJacobian]

method

newton_max_iter: int

pred_method

problem

simulate(x0: ndarray[tuple[Any, ...], dtype[float64]] | None = None, dx0: ndarray[tuple[Any, ...], dtype[float64]] | None = None, params0: ndarray[tuple[Any, ...], dtype[float64]] | None = None, boundary_updater: EmtBoundaryUpdateProtocol | None = None) → Tuple[ndarray[tuple[Any, ...], dtype[float64]], ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]], ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]], bool, bool]

Run the vectorized DAE time-domain simulation.

Parameters:

• x0 – Initial state vector.

• dx0 – Initial derivative vector.

• params0 – Initial event parameters.

• boundary_updater – Optional override for the problem boundary updater.

Returns:

Time vector and state trajectory matrix.

sorted_vars: List[Any]

t0

t_end

uid2idx_vars

vec_flat_args

vec_jacobian

vec_kernels

vectorized_ready

verbose

class VeraGridEngine.Simulations.EMT.solvers.StructuralVectorizedSolver.VectorizedKernelSpec(kernel: Callable[[...], Any], indices: ndarray, target_rows: ndarray)

Bases: object

Typed container for one vectorized residual kernel.

get_indices() → ndarray

Return the runtime gather matrix.

Returns:

Runtime gather matrix.

Return type:

np.ndarray

get_kernel() → Callable[[...], Any]

Return the compiled kernel.

Returns:

Compiled kernel.

Return type:

Callable[…, Any]

get_row_count() → int

Return the number of rows emitted by the kernel.

Returns:

Number of grouped rows.

Return type:

int

get_target_rows() → ndarray

Return the residual target rows.

Returns:

Residual target rows.

Return type:

np.ndarray

class VeraGridEngine.Simulations.EMT.solvers.StructuralVectorizedSolver.VectorizedResidualTrialEvaluator

Bases: object

Callable helper that evaluates vectorized residuals during backtracking.

set_context(residual_evaluator: Callable[[...], Any], full_params: ndarray[tuple[Any, ...], dtype[float64]], x_prev: ndarray[tuple[Any, ...], dtype[float64]], dx_prev: ndarray[tuple[Any, ...], dtype[float64]], h_eff: float, x_prev2: ndarray[tuple[Any, ...], dtype[float64]]) → None

Store the residual-evaluation context of the current Newton step.

Parameters:

• residual_evaluator (Callable[..., Any]) – Fixed-signature residual evaluator.

• full_params (Vec) – Full parameter vector.

• x_prev (Vec) – Previous accepted state.

• dx_prev (Vec) – Previous derivative vector.

• h_eff (float) – Effective local time step.

• x_prev2 (Vec) – Secondary state history.

Returns:

None.

Return type:

None

VeraGridEngine.Simulations.EMT.solvers.StructuralVectorizedSolver.build_residual_evaluator(residual_dispatcher: FusedResidualDispatcher | DirectResidualDispatcher, vec_flat_args: ndarray[tuple[Any, ...], dtype[float64]] | None = None) → Callable[[ndarray[tuple[Any, ...], dtype[float64]], ndarray[tuple[Any, ...], dtype[float64]], ndarray[tuple[Any, ...], dtype[float64]], ndarray[tuple[Any, ...], dtype[float64]], float, ndarray[tuple[Any, ...], dtype[float64]], ndarray[tuple[Any, ...], dtype[float64]]], None]

Build one fixed-signature residual evaluator for the current dispatcher.

The fused and direct dispatchers expose the same evaluate() API, so the Newton hot path can call one fixed-signature method without any dynamic attribute probing or signature branching.

Parameters:

• residual_dispatcher (FusedResidualDispatcher | DirectResidualDispatcher) – Residual dispatcher.

• vec_flat_args (Optional[Vec]) – Optional flat vectorized argument buffer used by legacy residual callables.

Returns:

Residual evaluator with signature (states, params, history, d_history, h, history2, out).

Return type:

Callable[[Vec, Vec, Vec, Vec, float, Vec, Vec], None]

VeraGridEngine.Simulations.EMT.solvers.StructuralVectorizedSolver.canonicalize_expression(expr: Expr, runtime_uids: Set[int], param_uids: Set[int]) → Tuple[str, List[Var]]

Returns a canonical structural signature and the ordered runtime variables only.

VeraGridEngine.Simulations.EMT.solvers.StructuralVectorizedSolver.evaluate_vectorized_residual(residual_evaluator: Callable[[...], Any], states: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]], history: ndarray[tuple[Any, ...], dtype[float64]], d_history: ndarray[tuple[Any, ...], dtype[float64]], h: float, history2: ndarray[tuple[Any, ...], dtype[float64]], residual_out: ndarray[tuple[Any, ...], dtype[float64]]) → float

Evaluate the fused vectorized residual into a caller-owned buffer.

Parameters:

• residual_evaluator (Callable[..., Any]) – Fixed-signature residual evaluator.

• states (Vec) – Current Newton iterate.

• params (Vec) – Full parameter vector.

• history (Vec) – Previous accepted state.

• d_history (Vec) – Previous derivative vector.

• h (float) – Effective local time step.

• history2 (Vec) – Secondary state history.

• residual_out (Vec) – Destination residual buffer.

Returns:

Residual infinity norm.

Return type:

float

VeraGridEngine.Simulations.EMT.solvers.StructuralVectorizedSolver.fill_full_parameter_buffer(runtime_params: ndarray[tuple[Any, ...], dtype[float64]], static_params: ndarray[tuple[Any, ...], dtype[float64]], full_params_out: ndarray[tuple[Any, ...], dtype[float64]]) → None

Write runtime and static parameters into one preallocated full buffer.

Parameters:

• runtime_params (Vec) – Runtime parameter slice.

• static_params (Vec) – Static parameter slice.

• full_params_out (Vec) – Destination full parameter vector.

Returns:

None.

Return type:

None

VeraGridEngine.Simulations.EMT.solvers.jit_symbolic_solver module

class VeraGridEngine.Simulations.EMT.solvers.jit_symbolic_solver.HybridJacobianEvaluator(kernel_jit: Callable, use_sparse: bool, n_vars: int, alpha_mult: float, parameter_count: int, np_rows: ndarray[tuple[Any, ...], dtype[int64]] | None = None, np_cols: ndarray[tuple[Any, ...], dtype[int64]] | None = None)

Bases: object

Evaluates the JIT-compiled Jacobian and formats it as either Dense or Sparse. This class replaces nested functions to ensure memory determinism and strict type safety.

alpha_mult

evaluate(states: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]], history: ndarray[tuple[Any, ...], dtype[float64]], d_history: ndarray[tuple[Any, ...], dtype[float64]], h: float, history2: ndarray[tuple[Any, ...], dtype[float64]] | None = None) → ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]] | csc_matrix

Executes the JIT kernel and constructs the resulting Jacobian matrix.

Parameters:

• states (Vec) – The current state vector.

• params (Vec) – The parameter vector.

• history (Vec) – The state history vector.

• d_history (Vec) – The derivative history vector.

• h (float) – The integration time step.

• history2 (Vec) – The secondary history vector (for BDF2).

Returns:

The assembled Jacobian matrix (Dense or Sparse CSC).

Return type:

Mat | CscMat

kernel_jit

n_vars

np_cols

np_rows

use_sparse

class VeraGridEngine.Simulations.EMT.solvers.jit_symbolic_solver.JitSymbolicSolver(problem: EmtProblemTemplate, t0: float, t_end: float, h: float, method: DynamicIntegrationMethod = DAE_Trapezoidal, pred_method: DynamicIntegrationMethod = None, dense_threshold: int = 100, verbose: bool = False, newton_max_iter: int = 15, newton_diag_config: NewtonDiagnosticsConfig | None = None, max_residual_inf_fail: float = inf, max_state_residual_inf_fail: float = inf)

Bases: object

algebraic_eqs

algebraic_vars

build_jit_kernel(method: DynamicIntegrationMethod) → None

Compiles the numerical residual function using JIT. It uses a batching strategy (splitting equations into chunks) to prevent Numba/LLVM from hanging when compiling very large non-linear systems.

Parameters:

method (DynamicIntegrationMethod) – The dynamic integration method to compile for.

Return type:

None

dense_threshold

diff_vars

dy: ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]]

get_backend_build_stats() → Dict[str, float]

Return setup timings collected during backend compilation.

Returns:

Backend build statistics.

Return type:

Dict[str, float]

get_last_runtime_stats() → Dict[str, Any]

Return runtime statistics collected during the latest simulation.

Returns:

Runtime statistics.

Return type:

Dict[str, Any]

get_last_sim_loop_time() → float

Return the latest integration-loop wall time.

Returns:

Latest loop time in seconds.

Return type:

float

h

jit_compiler

jit_jacobian_symbolic

jit_kernels

method

newton_max_iter: int

pred_method

problem

simulate(x0: ndarray[tuple[Any, ...], dtype[float64]] | None = None, dx0: ndarray[tuple[Any, ...], dtype[float64]] | None = None, params0: ndarray[tuple[Any, ...], dtype[float64]] | None = None, boundary_updater: EmtBoundaryUpdateProtocol | None = None) → Tuple[ndarray[tuple[Any, ...], dtype[float64]], ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]], ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]], bool, bool]

Main JIT simulation loop using the Symbolic Differentiation (SD) backend. :param x0: Initial state vector. :param dx0: Initial derivative vector. :param params0: Initial event parameters. :param boundary_updater: Optional override for the problem boundary updater. :return: A tuple containing the time vector and the state trajectory matrix. Tuple[Vec, Mat]

state_eqs

state_vars

steps

t: ndarray[tuple[Any, ...], dtype[float64]]

t0

t_end

verbose

y: ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]]

class VeraGridEngine.Simulations.EMT.solvers.jit_symbolic_solver.Predictor(n_states: int)

Bases: object

Computes an explicit predictor for the Newton initial guess x_iter.

n_states

predict(x_iter: ndarray[tuple[Any, ...], dtype[float64]], x_prev: ndarray[tuple[Any, ...], dtype[float64]], dx_prev: ndarray[tuple[Any, ...], dtype[float64]], h: float, pred_method: DynamicIntegrationMethod | None = DAE_BackEuler) → ndarray[tuple[Any, ...], dtype[float64]]

Apply predictor in-place and return x_iter. :param x_iter: Array to be written with the predictor guess (in-place). :param x_prev: Previous full variable vector at time n. :param dx_prev: Previous derivative vector (only first n_states are meaningful). :param h: Time step. :param pred_method: Integration method enum/value at the prediction step :return:

class VeraGridEngine.Simulations.EMT.solvers.jit_symbolic_solver.ResidualTrialEvaluator

Bases: object

Callable helper to evaluate residual norms during Newton backtracking.

dx_prev: ndarray[tuple[Any, ...], dtype[float64]] | None

full_params: ndarray[tuple[Any, ...], dtype[float64]] | None

h_eff: float

kernel_list: List[Callable[[...], Any]] | None

set_context(kernel_list: List[Callable[[...], Any]], full_params: ndarray[tuple[Any, ...], dtype[float64]], x_prev: ndarray[tuple[Any, ...], dtype[float64]], dx_prev: ndarray[tuple[Any, ...], dtype[float64]], h_eff: float, x_prev2: ndarray[tuple[Any, ...], dtype[float64]]) → None

Store the residual-evaluation context of the current Newton step.

Parameters:

• kernel_list (List[Callable[..., Any]]) – Batched residual kernels.

• full_params (Vec) – Full parameter vector.

• x_prev (Vec) – Previous accepted state.

• dx_prev (Vec) – Previous derivative vector.

• h_eff (float) – Effective local time step.

• x_prev2 (Vec) – Secondary state history.

Returns:

None.

Return type:

None

x_prev: ndarray[tuple[Any, ...], dtype[float64]] | None

x_prev2: ndarray[tuple[Any, ...], dtype[float64]] | None

class VeraGridEngine.Simulations.EMT.solvers.jit_symbolic_solver.SymbolicCompiledNumbaKernelCache

Bases: object

In-process cache of Numba-compiled symbolic kernels.

get(cache_key: str) → Callable[[...], Any] | None

Return one cached compiled kernel.

Parameters:

cache_key (str) – Deterministic cache key.

Returns:

Cached compiled kernel or None.

Return type:

Callable[…, Any] | None

set(cache_key: str, kernel: Callable[[...], Any]) → None

Store one compiled kernel.

Parameters:

• cache_key (str) – Deterministic cache key.

• kernel (Callable[..., Any]) – Compiled kernel.

Returns:

None.

Return type:

None

VeraGridEngine.Simulations.EMT.solvers.jit_symbolic_solver.evaluate_batched_residual(kernel_list: List[Callable[[...], Any]], x_iter: ndarray[tuple[Any, ...], dtype[float64]], full_params: ndarray[tuple[Any, ...], dtype[float64]], x_prev: ndarray[tuple[Any, ...], dtype[float64]], dx_prev: ndarray[tuple[Any, ...], dtype[float64]], h_eff: float, x_prev2: ndarray[tuple[Any, ...], dtype[float64]], residual_out: ndarray[tuple[Any, ...], dtype[float64]]) → float

Evaluate all residual batches into a caller-owned buffer.

Parameters:

• kernel_list (List[Callable[..., Any]]) – Batched residual kernels.

• x_iter (Vec) – Current Newton iterate.

• full_params (Vec) – Full parameter vector.

• x_prev (Vec) – Previous accepted state.

• dx_prev (Vec) – Previous derivative vector.

• h_eff (float) – Effective local time step.

• x_prev2 (Vec) – Secondary state history.

• residual_out (Vec) – Destination residual buffer.

Returns:

Residual infinity norm.

Return type:

float

VeraGridEngine.Simulations.EMT.solvers.jit_symbolic_solver.fill_full_parameter_buffer(runtime_params: ndarray[tuple[Any, ...], dtype[float64]], static_params: ndarray[tuple[Any, ...], dtype[float64]], full_params_out: ndarray[tuple[Any, ...], dtype[float64]]) → None

Write runtime and static parameters into one preallocated full buffer.

Parameters:

• runtime_params (Vec) – Runtime parameter slice.

• static_params (Vec) – Static parameter slice.

• full_params_out (Vec) – Destination full parameter vector.

Returns:

None.

Return type:

None

VeraGridEngine.Simulations.EMT.solvers.jit_symbolic_solver.get_vars_in_expr(expr: Expr) → Set[int]

Recursively collects UIDs of all variables present in an expression.

Parameters:

expr (Expr) – The symbolic expression to analyze.

Returns:

A set containing the unique identifiers (UIDs) of the variables.

Return type:

Set[int]

VeraGridEngine.Simulations.EMT.solvers.solver_AD module

class VeraGridEngine.Simulations.EMT.solvers.solver_AD.AdBacktrackingResidualEvaluator

Bases: object

Wrapper that evaluates AD residual batches for line-search backtracking.

evaluate(candidate_x: ndarray[tuple[Any, ...], dtype[float64]], out_res: ndarray[tuple[Any, ...], dtype[float64]]) → float

Evaluate one trial Newton iterate during backtracking.

Parameters:

• candidate_x (Vec) – Trial iterate.

• out_res (Vec) – Destination residual buffer.

Returns:

Residual infinity norm.

Return type:

float

set_context(kernel_list: List[Callable[[...], Any]], full_params: ndarray[tuple[Any, ...], dtype[float64]], x_prev: ndarray[tuple[Any, ...], dtype[float64]], dx_prev: ndarray[tuple[Any, ...], dtype[float64]], h_eff: float, x_prev2: ndarray[tuple[Any, ...], dtype[float64]]) → None

Store the residual-evaluation context of the current Newton step.

Parameters:

• kernel_list (List[Callable[..., Any]]) – Batched residual kernels.

• full_params (Vec) – Full parameter vector.

• x_prev (Vec) – Previous accepted state.

• dx_prev (Vec) – Previous derivative vector.

• h_eff (float) – Effective local time step.

• x_prev2 (Vec) – Secondary state history.

Returns:

None.

Return type:

None

class VeraGridEngine.Simulations.EMT.solvers.solver_AD.BoundaryUpdaterInterface

Bases: object

get_next_forced_event_time(t_prev: float, t_target: float) → float | None

update(t: float, x_prev: ndarray[tuple[Any, ...], dtype[float64]], full_params: ndarray[tuple[Any, ...], dtype[float64]]) → None

class VeraGridEngine.Simulations.EMT.solvers.solver_AD.JitAdSolver(problem: EmtProblemTemplate, t0: float, t_end: float, h: float, method: DynamicIntegrationMethod = DAE_Trapezoidal, pred_method: DynamicIntegrationMethod = None, dense_threshold: int = 100, verbose: bool = False, newton_max_iter: int = 15, newton_diag_config: NewtonDiagnosticsConfig | None = None)

Bases: object

algebraic_eqs

algebraic_vars

build_jit_ad(only_jacobian: bool = False) → None

Compiles the residual kernel using batching and prepares the Sparse AD Jacobian.

Parameters:

only_jacobian (bool) – If True, skips residual compilation (used by Vectorized Engine fallback).

Return type:

None

dense_threshold

dy: ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]]

get_backend_build_stats() → Dict[str, float]

Return setup timings collected during backend compilation.

Returns:

Backend build statistics.

Return type:

Dict[str, float]

get_last_runtime_stats() → Dict[str, float]

Return runtime statistics collected during the latest simulation.

Returns:

Runtime statistics.

Return type:

Dict[str, float]

get_last_sim_loop_time() → float

Return the latest integration-loop wall time.

Returns:

Latest loop time in seconds.

Return type:

float

h

jit_jacobian_ad: Dict[DynamicIntegrationMethod, SparseADJacobian]

jit_kernels_ad: Dict[DynamicIntegrationMethod, List[Callable]]

method

newton_max_iter: int

pred_method

problem

simulate(x0: ndarray[tuple[Any, ...], dtype[float64]] | None = None, dx0: ndarray[tuple[Any, ...], dtype[float64]] | None = None, params0: ndarray[tuple[Any, ...], dtype[float64]] | None = None, boundary_updater: EmtBoundaryUpdateProtocol | None = None) → Tuple[ndarray[tuple[Any, ...], dtype[float64]], ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]], ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]], bool, bool]

Main JIT simulation loop using the Automatic Differentiation (AD) backend.

Parameters:

• x0 – Initial state vector.

• dx0 – Initial derivative vector.

• params0 – Initial event parameters.

• boundary_updater – Optional override for the problem boundary updater.

Returns:

A tuple containing the time vector and state trajectory matrix.

state_eqs

state_vars

steps

t: ndarray[tuple[Any, ...], dtype[float64]]

t0

t_end

verbose

y: ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]]

class VeraGridEngine.Simulations.EMT.solvers.solver_AD.Predictor(n_states: int)

Bases: object

Computes an explicit predictor for the Newton initial guess x_iter.

n_states

predict(x_iter: ndarray[tuple[Any, ...], dtype[float64]], x_prev: ndarray[tuple[Any, ...], dtype[float64]], dx_prev: ndarray[tuple[Any, ...], dtype[float64]], h: float, pred_method: DynamicIntegrationMethod | None = DAE_BackEuler) → ndarray[tuple[Any, ...], dtype[float64]]

Apply predictor in-place and return x_iter. :param x_iter: Array to be written with the predictor guess (in-place). :param x_prev: Previous full variable vector at time n. :param dx_prev: Previous derivative vector (only first n_states are meaningful). :param h: Time step. :param pred_method: Integration method enum/value at the prediction step :return:

class VeraGridEngine.Simulations.EMT.solvers.solver_AD.SparseADJacobian(equations: ~typing.List[~typing.Any], variables: ~typing.List[~typing.Any], parameters: ~typing.List[~typing.Any], method: ~VeraGridEngine.enumerations.DynamicIntegrationMethod, use_cse: bool = True, dtype: ~typing.Any = <class 'numpy.float64'>)

Bases: object

Sparse Jacobian evaluator with DEBUG TIMING.

J

col_ptr

col_rows

color_cols

color_groups

color_ptr

colors

data_idx

dtype

equations

method

n_colors

n_cols

n_rows

param_map: Dict[int, int]

parameters

row_idx

use_cse

var_map: Dict[int, int]

variables

class VeraGridEngine.Simulations.EMT.solvers.solver_AD.SparseAdMasterJacobianDispatcher(ad_kernel: Callable[[...], Any], n_colors: int)

Bases: object

Callable dispatcher that executes one generic AD kernel across all colors.

VeraGridEngine.Simulations.EMT.solvers.solver_AD.evaluate_batched_residual(kernel_list: List[Callable[[...], Any]], x_iter: ndarray[tuple[Any, ...], dtype[float64]], full_params: ndarray[tuple[Any, ...], dtype[float64]], x_prev: ndarray[tuple[Any, ...], dtype[float64]], dx_prev: ndarray[tuple[Any, ...], dtype[float64]], h_eff: float, x_prev2: ndarray[tuple[Any, ...], dtype[float64]], residual_out: ndarray[tuple[Any, ...], dtype[float64]]) → float

Evaluate all residual batches into a caller-owned buffer.

Parameters:

• kernel_list (List[Callable[..., Any]]) – Batched residual kernels.

• x_iter (Vec) – Current Newton iterate.

• full_params (Vec) – Full parameter vector.

• x_prev (Vec) – Previous accepted state.

• dx_prev (Vec) – Previous derivative vector.

• h_eff (float) – Effective local time step.

• x_prev2 (Vec) – Secondary state history.

• residual_out (Vec) – Destination residual buffer.

Returns:

Residual infinity norm.

Return type:

float

VeraGridEngine.Simulations.EMT.solvers.solver_AD.fill_full_parameter_buffer(runtime_params: ndarray[tuple[Any, ...], dtype[float64]], static_params: ndarray[tuple[Any, ...], dtype[float64]], full_params_out: ndarray[tuple[Any, ...], dtype[float64]]) → None

Write runtime and static parameters into one preallocated full buffer.

Parameters:

• runtime_params (Vec) – Runtime parameter slice.

• static_params (Vec) – Static parameter slice.

• full_params_out (Vec) – Destination full parameter vector.

Returns:

None.

Return type:

None

VeraGridEngine.Simulations.EMT.solvers.solver_AD.greedy_color_columns(col_rows: List[List[int]], n_rows: int) → Tuple[ndarray, int]

Computes a greedy coloring of the column dependency graph to minimize AD sweeps.

Parameters:

• col_rows (List[List[int]]) – List of row indices where each column has non-zeros.

• n_rows (int) – Total number of rows in the matrix.

Returns:

An array of color IDs per column and the total number of colors.

Return type:

Tuple[np.ndarray, int]

VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver module

class VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver.CompiledNumbaKernelCache

Bases: object

In-process cache of Numba-compiled eager kernels.

get(cache_key: str) → Callable[[...], Any] | None

Return one cached compiled kernel.

Parameters:

cache_key (str) – Deterministic cache key.

Returns:

Cached compiled kernel or None.

Return type:

Callable[…, Any] | None

set(cache_key: str, kernel: Callable[[...], Any]) → None

Store one compiled kernel.

Parameters:

• cache_key (str) – Deterministic cache key.

• kernel (Callable[..., Any]) – Compiled kernel.

Returns:

None.

Return type:

None

class VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver.StructuralCompiledBoundaryUpdateWrapper

Bases: object

Boundary update interface for the structural compiled solver.

The structural compiled solver keeps the linear algebra outside Numba, but it still needs a typed boundary update object to refresh runtime parameters between local substeps.

get_next_forced_event_time(t_prev: float, t_target: float) → float | None

Return the next forced event time inside the local substep.

Parameters:

• t_prev (float) – Local substep start.

• t_target (float) – Local substep nominal end.

Returns:

Next forced event time or None.

Return type:

float | None

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

Update the parameter vector in place.

Parameters:

• t (float) – Current simulation time.

• x (Vec) – Current Newton iterate.

• params (Vec) – Full parameter vector.

Returns:

None

Return type:

None

class VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver.StructuralCompiledDirectResidualAssembler(kernel: Callable[[...], Any], equations: List[Expr], variables: List[Var], parameters: List[Var], method: DynamicIntegrationMethod, debug_info: List[Dict[str, Any]] | None)

Bases: object

Residual assembler backed by one monolithic in-place kernel.

evaluate(states: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]], history: ndarray[tuple[Any, ...], dtype[float64]], d_history: ndarray[tuple[Any, ...], dtype[float64]], h: float, history2: ndarray[tuple[Any, ...], dtype[float64]], data_out: ndarray[tuple[Any, ...], dtype[float64]]) → None

get_build_stats() → Dict[str, float]

get_work_buffer() → None

class VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver.StructuralCompiledEquationGroup

Bases: object

Structural group of equations that share the same vectorized template.

add_indices(variable_indices: List[int], row_index: int) → None

Append one grouped equation instance.

Parameters:

• variable_indices (List[int]) – Runtime indices used by this equation instance.

• row_index (int) – Global residual row index.

Returns:

None

Return type:

None

get_index_matrix() → List[List[int]]

Return the grouped runtime index matrix.

Returns:

Grouped runtime index matrix.

Return type:

List[List[int]]

get_row_indices() → List[int]

Return the residual rows covered by the group.

Returns:

Residual row indices.

Return type:

List[int]

get_template_eq() → Expr | None

Return the template equation.

Returns:

Template equation or None.

Return type:

Expr | None

get_template_vars() → List[Var] | None

Return the template runtime variables.

Returns:

Ordered template runtime variables or None.

Return type:

List[Var] | None

set_template(template_eq: Expr, template_vars: List[Var]) → None

Store the canonical template of the group.

Parameters:

• template_eq (Expr) – Canonical structural equation.

• template_vars (List[Var]) – Ordered runtime variables of the template.

Returns:

None

Return type:

None

class VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver.StructuralCompiledMasterJacobianDispatcher(ad_kernel: Callable[[...], Any], n_colors: int)

Bases: object

Callable dispatcher that evaluates structural AD colors into CSC storage.

class VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver.StructuralCompiledPredictor(n_states: int)

Bases: object

Explicit predictor used to initialize the Newton iteration.

predict(x_iter: ndarray[tuple[Any, ...], dtype[float64]], x_prev: ndarray[tuple[Any, ...], dtype[float64]], dx_prev: ndarray[tuple[Any, ...], dtype[float64]], h: float, pred_method: DynamicIntegrationMethod | None) → ndarray[tuple[Any, ...], dtype[float64]]

Write the predictor guess in place.

Parameters:

• x_iter (Vec) – Destination vector for the predictor guess.

• x_prev (Vec) – Previous variable vector.

• dx_prev (Vec) – Previous differential vector.

• h (float) – Local time step.

• pred_method (DynamicIntegrationMethod | None) – Predictor integration method.

Returns:

Updated predictor vector.

Return type:

Vec

class VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver.StructuralCompiledResidualAssembler(kernel_specs: List[StructuralCompiledVectorKernel], cache_token: str)

Bases: object

Residual dispatcher that reuses preallocated grouped work buffers.

evaluate(states: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]], history: ndarray[tuple[Any, ...], dtype[float64]], d_history: ndarray[tuple[Any, ...], dtype[float64]], h: float, history2: ndarray[tuple[Any, ...], dtype[float64]], data_out: ndarray[tuple[Any, ...], dtype[float64]]) → None

Evaluate the grouped residual system in place.

Parameters:

• states (Vec) – Current Newton iterate.

• params (Vec) – Full parameter vector.

• history (Vec) – Previous state vector.

• d_history (Vec) – Previous differential vector.

• h (float) – Local time step.

• history2 (Vec) – Secondary history vector.

• data_out (Vec) – Residual output buffer.

Returns:

None

Return type:

None

get_build_stats() → Dict[str, float]

Return residual assembler setup timings.

Returns:

Residual assembler build timings.

Return type:

Dict[str, float]

get_work_buffer() → ndarray[tuple[Any, ...], dtype[float64]]

Return the grouped work buffer.

Returns:

Grouped work buffer.

Return type:

Float64Matrix

class VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver.StructuralCompiledResidualDispatcher(kernel_specs: List[StructuralCompiledVectorKernel])

Bases: object

Residual dispatcher that orchestrates compiled group kernels explicitly.

The structural group kernels are already eagerly compiled with Numba. Keeping the final scatter loop in Python avoids fragile JIT/global binding behavior in the dispatcher layer and mirrors the residual assembly strategy used by the vectorized solver.

class VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver.StructuralCompiledResidualTrialEvaluator

Bases: object

Callable helper that evaluates residuals during Newton backtracking.

evaluate(trial_x: ndarray[tuple[Any, ...], dtype[float64]], out_res: ndarray[tuple[Any, ...], dtype[float64]]) → float

Evaluate one trial Newton iterate during backtracking.

Parameters:

• trial_x (Vec) – Trial iterate.

• out_res (Vec) – Destination residual buffer.

Returns:

Residual infinity norm.

Return type:

float

set_context(residual_assembler: StructuralCompiledResidualAssembler | StructuralCompiledDirectResidualAssembler, full_params: ndarray[tuple[Any, ...], dtype[float64]], x_prev: ndarray[tuple[Any, ...], dtype[float64]], dx_prev: ndarray[tuple[Any, ...], dtype[float64]], h_eff: float, x_prev2: ndarray[tuple[Any, ...], dtype[float64]]) → None

Store the residual-evaluation context of the current Newton step.

Parameters:

• residual_assembler (StructuralCompiledResidualAssembler | StructuralCompiledDirectResidualAssembler) – Residual assembler.

• full_params (Vec) – Full parameter vector.

• x_prev (Vec) – Previous accepted state.

• dx_prev (Vec) – Previous derivative vector.

• h_eff (float) – Effective local time step.

• x_prev2 (Vec) – Secondary state history.

Returns:

None

Return type:

None

class VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver.StructuralCompiledSolver(problem: EmtProblemTemplate, t0: float, t_end: float, h: float, method: DynamicIntegrationMethod = DAE_Trapezoidal, pred_method: DynamicIntegrationMethod | None = None, dense_threshold: int = 100, verbose: bool = False, newton_max_iter: int = 15, auto_build: bool = True, warmup_policy: StructuralCompiledWarmupPolicy = StructuralCompiledWarmupPolicy.Adaptive, sparse_solver_backend_provider: SparseLinearSolverBackendProvider | None = None, newton_diag_config: NewtonDiagnosticsConfig | None = None)

Bases: object

Standalone structural EMT solver with eager kernels and reusable buffers.

The solver keeps the linear algebra in SciPy, but it removes the Python-side allocation churn from residual assembly and sparse Jacobian evaluation.

get_backend_build_stats() → Dict[str, float]

Return setup timings for the structural compiled backend build.

Returns:

Setup timings grouped by build phase.

Return type:

Dict[str, float]

get_jacobian_data_buffer() → ndarray[tuple[Any, ...], dtype[float64]]

Return the persistent CSC numeric buffer.

Returns:

Persistent CSC numeric buffer.

Return type:

Vec

get_last_runtime_stats() → Dict[str, float]

Return runtime statistics collected during the latest simulation.

Returns:

Latest runtime statistics.

Return type:

Dict[str, float]

get_last_sim_loop_time() → float

Return the last measured integration loop wall time.

Returns:

Last measured loop time.

Return type:

float

get_residual_buffer() → ndarray[tuple[Any, ...], dtype[float64]]

Return the persistent residual buffer.

Returns:

Persistent residual buffer.

Return type:

Vec

is_vectorized_ready() → bool

Return whether the eager backend has already been built.

Returns:

True when the backend is ready.

Return type:

bool

simulate(x0: ndarray[tuple[Any, ...], dtype[float64]] | None = None, dx0: ndarray[tuple[Any, ...], dtype[float64]] | None = None, params0: ndarray[tuple[Any, ...], dtype[float64]] | None = None, boundary_updater: EmtBoundaryUpdateProtocol | None = None) → Tuple[ndarray[tuple[Any, ...], dtype[float64]], ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]], ndarray[tuple[Any, ...], dtype[float64]] | ndarray[tuple[int, int], dtype[float64]], bool, bool]

Run the implicit EMT simulation with eager structural kernels.

Parameters:

• x0 (Vec | None) – Optional initial state vector.

• dx0 (Vec | None) – Optional initial differential vector.

• params0 (Vec | None) – Optional initial runtime parameters.

• boundary_updater (EmtBoundaryUpdateProtocol | None) – Optional override for the problem boundary updater.

Returns:

Time vector, state trajectory and differential trajectory.

Return type:

Tuple[Vec, Mat, Mat]

class VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver.StructuralCompiledSparseADJacobian(equations: ~typing.List[~VeraGridEngine.Utils.Symbolic.symbolic.Expr], variables: ~typing.List[~VeraGridEngine.Utils.Symbolic.symbolic.Var], parameters: ~typing.List[~VeraGridEngine.Utils.Symbolic.symbolic.Var], method: ~VeraGridEngine.enumerations.DynamicIntegrationMethod, dtype: ~typing.Any = <class 'numpy.float64'>, use_cse: bool = True, eager_machine_code: bool = True, cache_token: str = 'default')

Bases: object

Sparse structural AD Jacobian that writes directly into CSC storage.

evaluate(states: ndarray[tuple[Any, ...], dtype[float64]], params: ndarray[tuple[Any, ...], dtype[float64]], history: ndarray[tuple[Any, ...], dtype[float64]], d_history: ndarray[tuple[Any, ...], dtype[float64]], h: float, history2: ndarray[tuple[Any, ...], dtype[float64]]) → csc_matrix

Evaluate the sparse Jacobian into the reusable CSC storage.

Parameters:

• states (Vec) – Current Newton iterate.

• params (Vec) – Full parameter vector.

• history (Vec) – Previous state vector.

• d_history (Vec) – Previous differential vector.

• h (float) – Local time step.

• history2 (Vec) – Secondary history vector.

Returns:

Reused CSC Jacobian matrix.

Return type:

csc_matrix

get_build_stats() → Dict[str, float]

Return sparse Jacobian build timings.

Returns:

Sparse Jacobian build timings.

Return type:

Dict[str, float]

get_data_buffer() → ndarray[tuple[Any, ...], dtype[float64]]

Return the reusable CSC numeric buffer.

Returns:

CSC numeric buffer.

Return type:

Vec

get_matrix() → csc_matrix

Return the reusable CSC matrix shell.

Returns:

CSC Jacobian matrix.

Return type:

csc_matrix

class VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver.StructuralCompiledSparseFactorizationManager(base_matrix: csc_matrix, base_data: ndarray[tuple[Any, ...], dtype[float64]], backend_provider: SparseLinearSolverBackendProvider | None = None)

Bases: object

Sparse linear-solve manager for StructuralCompiledSolver.

The manager keeps the current numeric factorization alive across time steps, caches one column ordering once it has been discovered, and reuses persistent numeric buffers so the solver hot loop does not rebuild sparse matrix shells.

factorize() → SparseLinearFactorizationHandle

Factorize the current Jacobian numeric values.

Returns:

Active sparse factorization handle.

Return type:

SparseLinearFactorizationHandle

get_active_matrix() → csc_matrix

Return the matrix used by the active numeric factorization path.

Returns:

Active sparse matrix.

Return type:

csc_matrix

get_stats() → Dict[str, float]

Return cumulative factorization-manager statistics.

Returns:

Cumulative statistics dictionary.

Return type:

Dict[str, float]

has_factorization() → bool

Return whether one numeric factorization is currently cached.

Returns:

True when a factorization is available.

Return type:

bool

invalidate() → None

Drop the current numeric factorization while keeping all persistent shells.

Returns:

None

Return type:

None

solve(rhs: ndarray[tuple[Any, ...], dtype[float64]]) → ndarray[tuple[Any, ...], dtype[float64]]

Solve the linear system using the current numeric factorization.

Parameters:

rhs (Vec) – Right-hand side vector.

Returns:

Solution buffer in original solver order.

Return type:

Vec

solve_fallback(rhs: ndarray[tuple[Any, ...], dtype[float64]]) → ndarray[tuple[Any, ...], dtype[float64]]

Solve the linear system with an iterative sparse fallback.

Parameters:

rhs (Vec) – Right-hand side vector.

Returns:

Solution buffer in original solver order.

Return type:

Vec

class VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver.StructuralCompiledVectorKernel(kernel: Callable[[...], Any], indices: ndarray[tuple[Any, ...], dtype[int32]], target_rows: ndarray[tuple[Any, ...], dtype[int32]])

Bases: object

Typed container for one vectorized residual kernel.

get_indices() → ndarray[tuple[Any, ...], dtype[int32]]

Return the runtime gather matrix.

Returns:

Runtime gather matrix.

Return type:

Int32Matrix

get_kernel() → Callable[[...], Any]

Return the compiled kernel.

Returns:

Compiled kernel.

Return type:

Callable[…, Any]

get_row_count() → int

Return the number of rows emitted by the kernel.

Returns:

Number of grouped rows.

Return type:

int

get_target_rows() → ndarray[tuple[Any, ...], dtype[int32]]

Return the residual target rows.

Returns:

Residual target rows.

Return type:

Int32Vector

class VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver.StructuralCompiledWarmupPolicy(value)

Bases: Enum

Warmup policy used by the structural compiled backend.

Adaptive = 'adaptive'

Full = 'full'

Off = 'off'

VeraGridEngine.Simulations.EMT.solvers.structural_compiled_solver.canonicalize_expression(expr: Expr, runtime_uids: Set[int], parameter_uids: Set[int]) → Tuple[str, List[Var]]

Return the canonical signature and ordered runtime variables of an expression.

Parameters:

• expr (Expr) – Symbolic expression to analyze.

• runtime_uids (Set[int]) – UIDs that belong to the runtime vector.

• parameter_uids (Set[int]) – UIDs that belong to the parameter vector.

Returns:

Canonical structural string and ordered runtime variables.

Return type:

Tuple[str, List[Var]]

Module contents