210 lines
7.0 KiB
Python
210 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script para verificar la extracción de resultados de TSNet
|
|
"""
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import json
|
|
|
|
# Configurar path para TSNet
|
|
sys.path.insert(
|
|
0,
|
|
r"D:\Proyectos\VisualStudio\CtrEditor\bin\Debug\net8.0-windows8.0\tsnet\Lib\site-packages",
|
|
)
|
|
|
|
try:
|
|
import tsnet
|
|
import wntr
|
|
import numpy as np
|
|
|
|
print("✅ TSNet y WNTR importados exitosamente")
|
|
except ImportError as e:
|
|
print("❌ Error importando librerías:", e)
|
|
sys.exit(1)
|
|
|
|
|
|
def test_results_extraction():
|
|
"""Test completo de extracción de resultados"""
|
|
|
|
print("\n🧪 INICIANDO TEST DE EXTRACCIÓN DE RESULTADOS")
|
|
print("=" * 60)
|
|
|
|
# Crear red de prueba simple
|
|
print("📋 Creando red de prueba...")
|
|
wn = wntr.network.WaterNetworkModel()
|
|
|
|
# Agregar nodos
|
|
wn.add_junction("J1", base_demand=0.0, elevation=0.0)
|
|
wn.add_junction("J2", base_demand=0.001, elevation=0.0) # 1 L/s
|
|
|
|
# Agregar reservorio
|
|
wn.add_reservoir("R1", base_head=50.0)
|
|
|
|
# Agregar tubería
|
|
wn.add_pipe("P1", "R1", "J1", length=100.0, diameter=0.1, roughness=0.001)
|
|
wn.add_pipe("P2", "J1", "J2", length=100.0, diameter=0.1, roughness=0.001)
|
|
|
|
print(
|
|
f"✅ Red creada: {len(wn.node_name_list)} nodos, {len(wn.pipe_name_list)} tuberías"
|
|
)
|
|
|
|
# Guardar red como archivo INP temporal y cargar con TSNet
|
|
print("💾 Guardando red como archivo INP...")
|
|
temp_inp = os.path.join(tempfile.gettempdir(), "test_network.inp")
|
|
wntr.network.write_inpfile(wn, temp_inp)
|
|
|
|
# Convertir a modelo transitorio
|
|
print("🔄 Convirtiendo a modelo transitorio...")
|
|
tm = tsnet.network.TransientModel(temp_inp)
|
|
|
|
# Configurar parámetros de simulación
|
|
tm.simulation_period = 2.0 # 2 segundos
|
|
tm.time_step = 0.1 # 0.1 segundos
|
|
|
|
print(f"✅ Configuración: {tm.simulation_period}s simulación, {tm.time_step}s paso")
|
|
|
|
# Aplicar correcciones automáticas (como en el código real)
|
|
print("🔧 Aplicando correcciones automáticas...")
|
|
corrections = 0
|
|
|
|
# Correcciones para nodos (junctions)
|
|
for node_name in tm.node_name_list:
|
|
node_obj = tm.get_node(node_name)
|
|
if hasattr(node_obj, "_demand") and not hasattr(node_obj, "demand_coeff"):
|
|
# Agregar demand_coeff si no existe
|
|
node_obj.demand_coeff = [1.0, 0.0, 0.0] # [a, b, c] para demanda variable
|
|
corrections += 1
|
|
|
|
# Correcciones para pipes
|
|
for pipe_name in tm.pipe_name_list:
|
|
pipe_obj = tm.get_link(pipe_name)
|
|
|
|
# Configurar atributos necesarios
|
|
pipe_length = getattr(pipe_obj, "length", 1.0)
|
|
num_segments = max(1, int(pipe_length / 0.1))
|
|
array_size = num_segments + 1
|
|
|
|
# Arrays iniciales
|
|
pipe_obj.initial_head = np.zeros(array_size)
|
|
pipe_obj.initial_velocity = np.zeros(array_size)
|
|
pipe_obj.wavev = 1000.0
|
|
pipe_obj.number_of_segments = num_segments
|
|
pipe_obj.roughness_height = getattr(pipe_obj, "roughness", 0.001)
|
|
|
|
# Arrays de resultados (usar arrays numpy)
|
|
num_time_steps = int(tm.simulation_period / tm.time_step) + 1
|
|
pipe_obj.start_node_velocity = np.zeros(num_time_steps)
|
|
pipe_obj.end_node_velocity = np.zeros(num_time_steps)
|
|
pipe_obj.start_node_head = np.zeros(num_time_steps)
|
|
pipe_obj.end_node_head = np.zeros(num_time_steps)
|
|
pipe_obj.start_node_flowrate = np.zeros(num_time_steps)
|
|
pipe_obj.end_node_flowrate = np.zeros(num_time_steps)
|
|
|
|
corrections += 10
|
|
print(
|
|
f" 📋 Pipe {pipe_name}: {num_segments} segmentos, arrays {array_size} elementos"
|
|
)
|
|
|
|
print(f"✅ {corrections} correcciones aplicadas")
|
|
|
|
# Ejecutar simulación
|
|
print("🚀 Ejecutando simulación TSNet...")
|
|
try:
|
|
results = tsnet.simulation.MOCSimulator(
|
|
tm, results_obj="results", friction="steady"
|
|
)
|
|
print("✅ Simulación TSNet completada exitosamente")
|
|
except Exception as e:
|
|
print(f"❌ Error en simulación: {e}")
|
|
return False
|
|
|
|
# EXTRACCIÓN DE RESULTADOS (como en el código real)
|
|
print("📊 Extrayendo resultados...")
|
|
|
|
node_results = {}
|
|
pipe_results = {}
|
|
|
|
# Extraer presiones de nodos
|
|
for node_name in tm.node_name_list:
|
|
try:
|
|
node_obj = tm.get_node(node_name)
|
|
if hasattr(node_obj, "head"):
|
|
if hasattr(node_obj.head, "__len__") and len(node_obj.head) > 0:
|
|
final_head = float(node_obj.head[-1])
|
|
node_results[node_name] = final_head
|
|
print(f" 🎯 Nodo {node_name}: {final_head:.3f} m")
|
|
else:
|
|
node_results[node_name] = 0.0
|
|
else:
|
|
node_results[node_name] = 0.0
|
|
except Exception as e:
|
|
print(f" ⚠️ Error en nodo {node_name}: {e}")
|
|
node_results[node_name] = 0.0
|
|
|
|
# Extraer flujos de tuberías
|
|
for pipe_name in tm.pipe_name_list:
|
|
try:
|
|
pipe_obj = tm.get_link(pipe_name)
|
|
if hasattr(pipe_obj, "start_node_flowrate"):
|
|
if (
|
|
hasattr(pipe_obj.start_node_flowrate, "__len__")
|
|
and len(pipe_obj.start_node_flowrate) > 0
|
|
):
|
|
final_flow = float(pipe_obj.start_node_flowrate[-1])
|
|
pipe_results[pipe_name] = final_flow
|
|
print(f" 💧 Pipe {pipe_name}: {final_flow:.6f} m³/s")
|
|
else:
|
|
pipe_results[pipe_name] = 0.0
|
|
else:
|
|
pipe_results[pipe_name] = 0.0
|
|
except Exception as e:
|
|
print(f" ⚠️ Error en pipe {pipe_name}: {e}")
|
|
pipe_results[pipe_name] = 0.0
|
|
|
|
# Crear estructura de resultados JSON
|
|
results_data = {
|
|
"nodes": node_results,
|
|
"pipes": pipe_results,
|
|
"simulation_time": tm.simulation_period,
|
|
"time_step": tm.time_step,
|
|
"extraction_test": True,
|
|
}
|
|
|
|
# Guardar resultados en archivo temporal
|
|
temp_dir = os.path.join(tempfile.gettempdir(), "TSNet")
|
|
os.makedirs(temp_dir, exist_ok=True)
|
|
results_file = os.path.join(temp_dir, "tsnet_results_test.json")
|
|
|
|
with open(results_file, "w") as f:
|
|
json.dump(results_data, f, indent=2)
|
|
|
|
print(f"💾 Resultados guardados en: {results_file}")
|
|
|
|
# Verificar archivo
|
|
if os.path.exists(results_file):
|
|
file_size = os.path.getsize(results_file)
|
|
print(f"✅ Archivo creado exitosamente ({file_size} bytes)")
|
|
|
|
# Leer y verificar contenido
|
|
with open(results_file, "r") as f:
|
|
loaded_data = json.load(f)
|
|
|
|
print(
|
|
f'📋 Verificación: {len(loaded_data["nodes"])} nodos, {len(loaded_data["pipes"])} pipes'
|
|
)
|
|
return True
|
|
else:
|
|
print("❌ Error: archivo no creado")
|
|
return False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
success = test_results_extraction()
|
|
print("\n" + "=" * 60)
|
|
if success:
|
|
print("🎉 TEST COMPLETADO EXITOSAMENTE - EXTRACCIÓN DE RESULTADOS FUNCIONANDO")
|
|
else:
|
|
print("❌ TEST FALLIDO - REVISAR IMPLEMENTACIÓN")
|
|
print("=" * 60)
|