Skip to content

Commit

Permalink
typing
Browse files Browse the repository at this point in the history 
 - contingency_analysis_driver.py
 - power_flow_worker.py
  • Loading branch information
@ramferan
ramferan committed Jun 21, 2023
1 parent f0a2d0b commit 44ac134
Show file tree
Hide file tree
Showing 2 changed files with 69 additions and 40 deletions.
40 changes: 26 additions & 14 deletions src/GridCal/Engine/Simulations/ContingencyAnalysis/contingency_analysis_driver.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -138,32 +138,40 @@ def n_minus_k(self, t=None):
numerical_circuit = compile_numerical_circuit_at(self.grid, t_idx=t)

if self.options.pf_options is None:
pf_opts = PowerFlowOptions(solver_type=SolverType.DC,
ignore_single_node_islands=True)
pf_opts = PowerFlowOptions(
solver_type=SolverType.DC,
ignore_single_node_islands=True
)

else:
pf_opts = self.options.pf_options

# declare the results
results = ContingencyAnalysisResults(ncon=len(self.grid.contingency_groups),
nbr=numerical_circuit.nbr,
nbus=numerical_circuit.nbus,
branch_names=numerical_circuit.branch_names,
bus_names=numerical_circuit.bus_names,
bus_types=numerical_circuit.bus_types,
con_names=self.grid.get_contingency_group_names())
results = ContingencyAnalysisResults(
ncon=len(self.grid.contingency_groups),
nbr=numerical_circuit.nbr,
nbus=numerical_circuit.nbus,
branch_names=numerical_circuit.branch_names,
bus_names=numerical_circuit.bus_names,
bus_types=numerical_circuit.bus_types,
con_names=self.grid.get_contingency_group_names()
)

# get contingency groups dictionary
cg_dict = self.grid.get_contingency_group_dict()

branches_dict = {e.idtag: ei for ei, e in enumerate(self.grid.get_branches_wo_hvdc())}
branches_dict = self.grid.get_branches_wo_hvdc_dict()

# keep the original states
original_br_active = numerical_circuit.branch_data.active.copy()
original_gen_active = numerical_circuit.generator_data.active.copy()
original_gen_p = numerical_circuit.generator_data.p.copy()

# run 0
pf_res_0 = multi_island_pf_nc(nc=numerical_circuit, options=pf_opts)
pf_res_0 = multi_island_pf_nc(
nc=numerical_circuit,
options=pf_opts
)

# for each contingency group
for ic, contingency_group in enumerate(self.grid.contingency_groups):
Expand All @@ -181,17 +189,21 @@ def n_minus_k(self, t=None):
if cnt.prop == 'active':
numerical_circuit.branch_data.active[br_idx] = int(cnt.value)
else:
print('Unknown contingency property ', cnt.prop, 'at', cnt.name, cnt.idtag)
print(f'Unknown contingency property {cnt.prop} at {cnt.name} {cnt.idtag}')
else:
pass

# report progress
if t is None:
self.progress_text.emit('Contingency group:' + contingency_group.name)
self.progress_text.emit(f'Contingency group: {contingency_group.name}')
self.progress_signal.emit((ic + 1) / len(self.grid.contingency_groups) * 100)

# run
pf_res = multi_island_pf_nc(nc=numerical_circuit, options=pf_opts, V_guess=pf_res_0.voltage)
pf_res = multi_island_pf_nc(
nc=numerical_circuit,
options=pf_opts,
V_guess=pf_res_0.voltage
)

results.Sf[ic, :] = pf_res.Sf
results.S[ic, :] = pf_res.Sbus
Expand Down
69 changes: 43 additions & 26 deletions src/GridCal/Engine/Simulations/PowerFlow/power_flow_worker.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
from GridCal.Engine.Simulations.PowerFlow.power_flow_results import PowerFlowResults
from GridCal.Engine.Simulations.PowerFlow.power_flow_options import PowerFlowOptions
from GridCal.Engine.Simulations.PowerFlow.power_flow_results import NumericPowerFlowResults
from GridCal.Engine.Simulations.OPF.opf_results import OptimalPowerFlowResults
from GridCal.Engine.Core.numerical_circuit import NumericalCircuit
from GridCal.Engine.Core.multi_circuit import MultiCircuit
from GridCal.Engine.Core.numerical_circuit import compile_numerical_circuit_at
Expand Down Expand Up @@ -590,28 +591,37 @@ def get_hvdc_power(multi_circuit: MultiCircuit, bus_dict, theta, t=None):
return Shvdc, Losses_hvdc, Pf_hvdc, Pt_hvdc, loading_hvdc, n_free


def multi_island_pf_nc(nc: NumericalCircuit, options: PowerFlowOptions, logger=bs.Logger(),
V_guess=None) -> "PowerFlowResults":
def multi_island_pf_nc(
nc: NumericalCircuit,
options: PowerFlowOptions,
logger=bs.Logger(),
V_guess=None
) -> "PowerFlowResults":
"""
Multiple islands power flow (this is the most generic power flow function)
:param nc: SnapshotData instance
:param options: PowerFlowOptions instance
:param logger: logger
:param V_guess:
:return: PowerFlowResults instance
"""

# declare results
results = PowerFlowResults(n=nc.nbus,
m=nc.nbr,
n_hvdc=nc.nhvdc,
bus_names=nc.bus_data.names,
branch_names=nc.branch_data.names,
hvdc_names=nc.hvdc_data.names,
bus_types=nc.bus_data.bus_types)
results = PowerFlowResults(
n=nc.nbus,
m=nc.nbr,
n_hvdc=nc.nhvdc,
bus_names=nc.bus_data.names,
branch_names=nc.branch_data.names,
hvdc_names=nc.hvdc_data.names,
bus_types=nc.bus_data.bus_types,
)

# compose the HVDC power injections
Shvdc, Losses_hvdc, Pf_hvdc, Pt_hvdc, loading_hvdc, n_free = nc.hvdc_data.get_power(Sbase=nc.Sbase,
theta=np.zeros(nc.nbus))
Shvdc, Losses_hvdc, Pf_hvdc, Pt_hvdc, loading_hvdc, n_free = nc.hvdc_data.get_power(
Sbase=nc.Sbase,
theta=np.zeros(nc.nbus),
)

# remember the initial hvdc control values
Losses_hvdc_prev = Losses_hvdc.copy()
Expand All @@ -621,7 +631,9 @@ def multi_island_pf_nc(nc: NumericalCircuit, options: PowerFlowOptions, logger=b
Shvdc_prev = Shvdc.copy()

# compute islands
islands = nc.split_into_islands(ignore_single_node_islands=options.ignore_single_node_islands)
islands = nc.split_into_islands(
ignore_single_node_islands=options.ignore_single_node_islands,
)

# initialize the all controls var
all_controls_ok = False # to run the first time
Expand Down Expand Up @@ -655,23 +667,26 @@ def multi_island_pf_nc(nc: NumericalCircuit, options: PowerFlowOptions, logger=b
Qmin=island.Qmin_bus,
Qmax=island.Qmax_bus,
options=options,
logger=logger
logger=logger,
)

# merge the results from this island
results.apply_from_island(results=res,
b_idx=island.original_bus_idx,
br_idx=island.original_branch_idx)
results.apply_from_island(
results=res,
b_idx=island.original_bus_idx,
br_idx=island.original_branch_idx,
)

else:
logger.add_info('No slack nodes in the island', str(i))
# --------------------------------------------------------------------------------------------------------------

if n_free and control_iter < max_control_iter:

Shvdc, Losses_hvdc, Pf_hvdc, Pt_hvdc, loading_hvdc, n_free = nc.hvdc_data.get_power(Sbase=nc.Sbase,
theta=np.angle(
results.voltage))
Shvdc, Losses_hvdc, Pf_hvdc, Pt_hvdc, loading_hvdc, n_free = nc.hvdc_data.get_power(
Sbase=nc.Sbase,
theta=np.angle(results.voltage),
)

# hvdc_control_err = np.max(np.abs(Pf_hvdc_prev - Pf_hvdc))
hvdc_control_err = np.max(np.abs(Shvdc - Shvdc_prev))
Expand Down Expand Up @@ -729,13 +744,15 @@ def multi_island_pf_nc(nc: NumericalCircuit, options: PowerFlowOptions, logger=b
return results


def multi_island_pf(multi_circuit: MultiCircuit,
options: PowerFlowOptions,
opf_results: Union["OptimalPowerFlowResults", None] = None,
t: Union[int, None] = None,
logger: bs.Logger = bs.Logger(),
bus_dict: Union[Dict[Bus, int], None] = None,
areas_dict: Union[Dict[Area, int], None] = None) -> "PowerFlowResults":
def multi_island_pf(
multi_circuit: MultiCircuit,
options: PowerFlowOptions,
opf_results: Union[OptimalPowerFlowResults, None] = None,
t: Union[int, None] = None,
logger: bs.Logger = bs.Logger(),
bus_dict: Union[Dict[Bus, int], None] = None,
areas_dict: Union[Dict[Area, int], None] = None
) -> "PowerFlowResults":
"""
Multiple islands power flow (this is the most generic power flow function)
:param multi_circuit: MultiCircuit instance
Expand Down

0 comments on commit 44ac134

Please sign in to comment.