#!/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)