"""Network Calculation Threads Module
===================================
Threaded network initialization and calculation functionality.
:author: Dipl.-Ing. (FH) Jonas Pfeiffer
"""
import traceback
import numpy as np
from PyQt6.QtCore import QThread, pyqtSignal
from districtheatingsim.net_simulation_pandapipes.pp_net_initialisation_geojson import initialize_geojson
from districtheatingsim.net_simulation_pandapipes.pp_net_time_series_simulation import thermohydraulic_time_series_net, time_series_preprocessing, simplified_time_series_net
from districtheatingsim.net_simulation_pandapipes.utilities import optimize_diameter_types
[docs]
class NetInitializationThread(QThread):
"""
Thread for network initialization tasks.
"""
calculation_done = pyqtSignal(object)
calculation_error = pyqtSignal(str)
[docs]
def __init__(self, NetworkGenerationData):
"""
Initialize network initialization thread.
:param NetworkGenerationData: Network generation data object.
:type NetworkGenerationData: object
"""
super().__init__()
self.NetworkGenerationData = NetworkGenerationData
[docs]
def run(self):
"""
Run network initialization process.
"""
try:
self.NetworkGenerationData = initialize_geojson(self.NetworkGenerationData)
# Diameter optimization if enabled
if self.NetworkGenerationData.diameter_optimization_pipe_checked == True:
self.NetworkGenerationData.net = optimize_diameter_types(self.NetworkGenerationData.net, self.NetworkGenerationData.max_velocity_pipe, self.NetworkGenerationData.material_filter_pipe, self.NetworkGenerationData.k_mm_pipe)
self.calculation_done.emit(self.NetworkGenerationData)
except Exception as e:
self.calculation_error.emit(str(e) + "\n" + traceback.format_exc())
[docs]
def stop(self):
"""Stop thread execution."""
if self.isRunning():
self.requestInterruption()
self.wait()
[docs]
class NetCalculationThread(QThread):
"""
Thread for network time series calculations.
"""
calculation_done = pyqtSignal(object)
calculation_error = pyqtSignal(str)
[docs]
def __init__(self, NetworkGenerationData, simplified=False):
"""
Initialize calculation thread.
:param NetworkGenerationData: Network generation data object.
:type NetworkGenerationData: object
:param simplified: Use simplified fast calculation instead of detailed simulation.
:type simplified: bool
"""
super().__init__()
self.NetworkGenerationData = NetworkGenerationData
self.simplified = simplified
[docs]
def run(self):
"""
Run time series calculation process.
"""
try:
self.NetworkGenerationData = time_series_preprocessing(self.NetworkGenerationData)
if self.simplified:
# Use simplified fast calculation
self.NetworkGenerationData = simplified_time_series_net(self.NetworkGenerationData)
else:
# Use detailed hydraulic simulation
self.NetworkGenerationData = thermohydraulic_time_series_net(self.NetworkGenerationData)
self.calculation_done.emit(self.NetworkGenerationData)
except Exception as e:
self.calculation_error.emit(str(e) + "\n" + traceback.format_exc())
[docs]
def stop(self):
"""Stop thread execution."""
if self.isRunning():
self.requestInterruption()
self.wait()