# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
# SPDX-License-Identifier: MPL-2.0
from __future__ import annotations
import time
import numpy as np
from typing import Union, TYPE_CHECKING
from VeraGridEngine.basic_structures import IntVec, Vec
from VeraGridEngine.basic_structures import Logger, Mat
from VeraGridEngine.enumerations import EngineType, SimulationTypes
from VeraGridEngine.Devices.multi_circuit import MultiCircuit
from VeraGridEngine.Simulations.results_template import ResultsTemplate
if TYPE_CHECKING:
from VeraGridEngine.Simulations.Clustering.clustering_results import ClusteringResults
[docs]
class DummySignal:
"""
Qt signal placeholder to not to import QT in the engine
"""
__slots__ = ("tpe",)
def __init__(self, tpe: type = str) -> None:
self.tpe = tpe
[docs]
def emit(self, val: Union[str, float] = '') -> None:
pass
[docs]
def connect(self, val):
"""
:param val:
"""
pass
[docs]
class DriverToSave:
"""
Wrapper to save a driver
"""
__slots__ = (
"name",
"tpe",
"results",
"logger",
)
def __init__(self,
name: str,
tpe: SimulationTypes,
results: ResultsTemplate,
logger: Logger):
"""
:param name:
:param tpe:
:param results:
:param logger:
"""
self.name = name
self.tpe: SimulationTypes = tpe
self.results: ResultsTemplate = results
self.logger: Logger = logger
[docs]
class DriverTemplate:
"""
Base driver template
"""
__slots__ = (
"progress_signal",
"progress_text",
"done_signal",
"grid",
"results",
"engine",
"elapsed",
"logger",
"__cancel__",
"_is_running",
"_DriverTemplate__start",
)
tpe = SimulationTypes.TemplateDriver
name = 'Template'
def __init__(self,
grid: MultiCircuit,
engine: EngineType = EngineType.VeraGrid):
"""
Constructor
:param grid: MultiCircuit instance
:param engine: EngineType
"""
self.progress_signal = DummySignal()
self.progress_text = DummySignal(str)
self.done_signal = DummySignal()
self.grid: MultiCircuit = grid
self.results = None
self.engine = engine
self.elapsed = 0
self.logger = Logger()
self.__cancel__ = False
self._is_running = False
self.__start = time.time()
[docs]
def tic(self, skip_logger=False):
"""
Register start of time
"""
self.__start = time.time()
if not skip_logger:
self.logger.add_info(msg="Elapsed total (s)",
device_property="Started")
[docs]
def toc(self, skip_logger=False):
"""
Register end of time
:param skip_logger: skip logging this?
"""
self.elapsed = time.time() - self.__start
if not skip_logger:
self.logger.add_info(msg="Elapsed total (s)",
device_property="Ended (s)",
value='{:.4f}'.format(self.elapsed))
[docs]
def get_steps(self):
"""
Get the number of steps in the simulation
:return:
"""
return list()
[docs]
def run(self):
"""
"""
pass
[docs]
def copy_signals(self, other: "TimeSeriesDriverTemplate"):
"""
Copy the signals from another driver
:param other:
:return:
"""
self.progress_signal = other.progress_signal
self.progress_text = other.progress_text
self.done_signal = other.done_signal
[docs]
def report_progress(self, val: float):
"""
Report progress
:param val: float value
"""
self.progress_signal.emit(val)
[docs]
def report_progress2(self, current: int, total: int):
"""
Report progress
:param current: current value (zero based)
:param total: total value
"""
val = ((current + 1) / total) * 100
self.progress_signal.emit(val)
[docs]
def report_done(self, txt="done!", val=0.0):
"""
Report done
"""
self.report_progress(val)
self.report_text(txt)
self.done_signal.emit()
[docs]
def report_text(self, val: str):
"""
Report text
:param val: text value
"""
self.progress_text.emit(val)
[docs]
def cancel(self):
"""
Cancel the simulation
"""
self.__cancel__ = True
self.report_done("Cancelled!")
[docs]
def is_cancel(self) -> bool:
"""
Check if cancel was activated
:return:
"""
return self.__cancel__
[docs]
def isRunning(self):
"""
:return:
"""
return self._is_running
[docs]
def get_save_data(self) -> DriverToSave:
"""
Get save data representation of this driver
:return: DriverToSave
"""
return DriverToSave(name=self.name,
tpe=self.tpe,
results=self.results,
logger=self.logger)
[docs]
class TimeSeriesDriverTemplate(DriverTemplate):
"""
Time series driver template
"""
__slots__ = (
"clustering_results",
"using_clusters",
"time_indices",
"sampled_probabilities",
"original_sample_idx",
)
def __init__(
self,
grid: MultiCircuit,
time_indices: IntVec | None,
clustering_results: Union[ClusteringResults, None] = None,
engine: EngineType = EngineType.VeraGrid,
check_time_series: bool = True):
"""
Time Series driver constructor
:param grid: MultiCircuit instance
:param time_indices: array of time indices to simulate (optional)
:param clustering_results: ClusteringResults object (optional)
"""
if not grid.has_time_series and check_time_series:
raise Exception(self.name + " can only run in grids with time series data :(")
DriverTemplate.__init__(self, grid=grid, engine=engine)
self.clustering_results: Union[ClusteringResults, None] = clustering_results
if clustering_results:
self.using_clusters = True
self.time_indices: IntVec = clustering_results.time_indices
self.sampled_probabilities: Vec = clustering_results.sampled_probabilities
self.original_sample_idx: IntVec = clustering_results.original_sample_idx
else:
self.using_clusters = False
self.original_sample_idx = None
if time_indices is None:
self.time_indices = None
self.sampled_probabilities = None
else:
if len(time_indices) == 0:
self.time_indices = None
self.sampled_probabilities = None
else:
self.time_indices: IntVec = time_indices
self.sampled_probabilities: Vec = np.ones(shape=len(self.time_indices)) / len(self.time_indices)
[docs]
def get_steps(self):
"""
Get time steps list of strings
"""
if self.time_indices is None:
return []
else:
return [self.grid.time_profile[i].strftime('%d-%m-%Y %H:%M') for i in self.time_indices]
[docs]
def get_fuel_emissions_energy_calculations(self, gen_p: Mat, gen_cost: Mat):
"""
Calculate fuel emissions and energy cost
:param gen_p:
:param gen_cost:
:return:
"""
# gather the fuels and emission rates matrices
gen_emissions_rates_matrix = self.grid.get_gen_emission_rates_sparse_matrix()
gen_fuel_rates_matrix = self.grid.get_gen_fuel_rates_sparse_matrix()
system_fuel = (gen_fuel_rates_matrix * gen_p.T).T
system_emissions = (gen_emissions_rates_matrix * gen_p.T).T
with np.errstate(divide='ignore', invalid='ignore'): # numpy magic to ignore the zero divisions
system_energy_cost = np.nan_to_num(gen_cost / gen_p).sum(axis=1)
return system_fuel, system_emissions, system_energy_cost