Remove outdated hydraulic system test scripts and results
- Deleted HydraulicSystemTests_TimingV2.py as it is no longer needed. - Removed previous test results file HydraulicTestResults_20250106.json. - Deleted timing test results file HydraulicTestResults_TimingV2_20250906_203611.json. - Removed summary markdown file ResumenPruebasHidraulicas.md as it is obsolete. - Deleted batch script start_mcp_proxy.bat for starting the MCP proxy. - Removed Python script start_mcp_proxy.py which initiated the MCP proxy. - Deleted PowerShell script update_single_file.ps1 used for updating methods in files.
This commit is contained in:
parent
5c03e19207
commit
c76e4c1749
|
|
@ -57,6 +57,15 @@ namespace CtrEditor
|
|||
private const double MAX_PLC_INTERVAL = 100; // Máximo intervalo PLC (ms)
|
||||
private const double SIM_PRIORITY_TIME = 10; // Tiempo reservado para simulación (ms)
|
||||
|
||||
// Sistema adaptativo para timing de simulación
|
||||
private double lastSimExecutionTime = 0;
|
||||
private double maxSimExecutionTime = 0;
|
||||
private int simTimingAdaptationCounter = 0;
|
||||
private const int SIM_ADAPTATION_SAMPLES = 5; // Evaluar cada 5 muestras para más responsividad
|
||||
private const double MIN_SIM_INTERVAL = 8; // Mínimo intervalo simulación (ms)
|
||||
private const double MAX_SIM_INTERVAL = 100; // Máximo intervalo simulación (ms)
|
||||
private const double SIM_BUFFER_TIME = 2; // Buffer de 2ms extra respecto al tiempo real
|
||||
|
||||
private float TiempoDesdeStartSimulacion;
|
||||
private bool Debug_SimulacionCreado = false;
|
||||
|
||||
|
|
@ -414,7 +423,7 @@ namespace CtrEditor
|
|||
ItemDoubleClickCommand = new ParameterizedRelayCommand(ExecuteDoubleClick);
|
||||
|
||||
_timerSimulacion = new System.Timers.Timer();
|
||||
_timerSimulacion.Interval = 10; // 10ms - más preciso que DispatcherTimer
|
||||
_timerSimulacion.Interval = 15; // 15ms inicial - más conservador para permitir adaptación
|
||||
_timerSimulacion.Elapsed += OnTickSimulacion;
|
||||
_timerSimulacion.AutoReset = true; // Reinicio automático
|
||||
_timerSimulacion.SynchronizingObject = null; // No sincronizar automáticamente con UI
|
||||
|
|
@ -1122,6 +1131,9 @@ namespace CtrEditor
|
|||
|
||||
IsSimulationRunning = true;
|
||||
|
||||
// Resetear contadores adaptativos para timing limpio
|
||||
ResetAdaptiveTimingCounters();
|
||||
|
||||
// Ocultar rectángulos de selección antes de iniciar la simulación
|
||||
_objectManager.UpdateSelectionVisuals();
|
||||
|
||||
|
|
@ -1153,6 +1165,9 @@ namespace CtrEditor
|
|||
_timerSimulacion.Stop();
|
||||
_timerPLCUpdate.Stop(); // También detener el timer PLC
|
||||
|
||||
// Resetear contadores adaptativos al detener
|
||||
ResetAdaptiveTimingCounters();
|
||||
|
||||
// Restaurar los rectángulos de selección si hay objetos seleccionados
|
||||
_objectManager.UpdateSelectionVisuals();
|
||||
|
||||
|
|
@ -1212,6 +1227,10 @@ namespace CtrEditor
|
|||
}
|
||||
|
||||
executionStopwatch.Stop(); // ✅ CORREGIDO: Detener cronómetro de ejecución
|
||||
|
||||
// Sistema adaptativo de timing de simulación
|
||||
AdaptSimulationTiming(executionStopwatch.Elapsed.TotalMilliseconds);
|
||||
|
||||
//Debug.WriteLine($"OnTickSimulacion execution time: {stopwatch.ElapsedMilliseconds} ms");
|
||||
//Debug.WriteLine($"OnTickSimulacion execution time: {executionStopwatch.Elapsed.TotalMilliseconds:F2}ms | Timer interval: {timeBetweenCalls:F2}ms | Objects: {ObjetosSimulables?.Count ?? 0}");
|
||||
});
|
||||
|
|
@ -1352,6 +1371,73 @@ namespace CtrEditor
|
|||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sistema adaptativo para el timing de simulación que mantiene 2ms de buffer extra
|
||||
/// </summary>
|
||||
private void AdaptSimulationTiming(double executionTime)
|
||||
{
|
||||
lastSimExecutionTime = executionTime;
|
||||
maxSimExecutionTime = Math.Max(maxSimExecutionTime, executionTime);
|
||||
simTimingAdaptationCounter++;
|
||||
|
||||
// Evaluar cada SIM_ADAPTATION_SAMPLES ejecuciones para responsividad rápida
|
||||
if (simTimingAdaptationCounter >= SIM_ADAPTATION_SAMPLES)
|
||||
{
|
||||
double currentInterval = _timerSimulacion.Interval;
|
||||
double newInterval = currentInterval;
|
||||
|
||||
// Calcular intervalo ideal: tiempo de ejecución máximo + buffer de 2ms
|
||||
double idealInterval = maxSimExecutionTime + SIM_BUFFER_TIME;
|
||||
|
||||
// Si el tiempo de ejecución + buffer > intervalo actual, necesitamos más tiempo
|
||||
if (idealInterval > currentInterval)
|
||||
{
|
||||
// Incrementar intervalo para evitar que el sistema se vuelva inusable
|
||||
newInterval = Math.Min(idealInterval + 1, MAX_SIM_INTERVAL); // +1ms adicional de seguridad
|
||||
Debug.WriteLine($"Simulación Timer: Aumentando intervalo a {newInterval:F1}ms (ejecución máxima: {maxSimExecutionTime:F1}ms)");
|
||||
}
|
||||
// Si hemos estado muy por debajo del umbral, podemos reducir el intervalo
|
||||
else if (idealInterval < currentInterval * 0.8)
|
||||
{
|
||||
newInterval = Math.Max(idealInterval, MIN_SIM_INTERVAL);
|
||||
Debug.WriteLine($"Simulación Timer: Reduciendo intervalo a {newInterval:F1}ms (ejecución máxima: {maxSimExecutionTime:F1}ms)");
|
||||
}
|
||||
|
||||
// Aplicar el nuevo intervalo si cambió significativamente
|
||||
if (Math.Abs(newInterval - currentInterval) > 0.5)
|
||||
{
|
||||
_timerSimulacion.Interval = newInterval;
|
||||
Debug.WriteLine($"Simulación Timer: Intervalo actualizado de {currentInterval:F1}ms a {newInterval:F1}ms");
|
||||
}
|
||||
|
||||
// Reset para el próximo período de evaluación
|
||||
maxSimExecutionTime = 0;
|
||||
simTimingAdaptationCounter = 0;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Resetea todos los contadores adaptativos de timing para empezar con mediciones limpias
|
||||
/// </summary>
|
||||
private void ResetAdaptiveTimingCounters()
|
||||
{
|
||||
// Reset contadores de simulación
|
||||
lastSimExecutionTime = 0;
|
||||
maxSimExecutionTime = 0;
|
||||
simTimingAdaptationCounter = 0;
|
||||
|
||||
// Reset contadores de PLC
|
||||
lastPlcExecutionTime = 0;
|
||||
maxPlcExecutionTime = 0;
|
||||
plcTimingAdaptationCounter = 0;
|
||||
|
||||
// Restablecer intervalos iniciales conservadores
|
||||
_timerSimulacion.Interval = 15; // 15ms inicial
|
||||
_timerPLCUpdate.Interval = 10; // 10ms inicial
|
||||
|
||||
Debug.WriteLine("Timing adaptativo: Contadores y intervalos reseteados");
|
||||
}
|
||||
|
||||
private void OpenWorkDirectory()
|
||||
{
|
||||
var dialog = new VistaFolderBrowserDialog();
|
||||
|
|
|
|||
|
|
@ -1,138 +0,0 @@
|
|||
# Script para estandarizar objetos derivados de osBase
|
||||
# Ejecutar desde la raíz del proyecto CtrEditor
|
||||
|
||||
Write-Host "=== Estandarización de Objetos osBase ===" -ForegroundColor Green
|
||||
|
||||
# Mapeo de nombres de clase mejorados
|
||||
$nombresClase = @{
|
||||
# Transportes
|
||||
"ucTransporteTTop" = "Transporte TTOP"
|
||||
"ucTransporteTTopDualInverter" = "Transporte TTOP Dual Inverter"
|
||||
"ucTransporteGuias" = "Transporte con Guías"
|
||||
"ucTransporteGuiasUnion" = "Transporte con Guías Unión"
|
||||
"ucTransporteCurva" = "Transporte Curva"
|
||||
"ucTransporteCurvaGuias" = "Transporte Curva con Guías"
|
||||
|
||||
# Sensores
|
||||
"ucPhotocell" = "Fotocélula"
|
||||
"ucGearEncoder" = "Encoder de Engranaje"
|
||||
"ucEncoderMotor" = "Encoder Motor"
|
||||
"ucEncoderMotorLineal" = "Encoder Motor Lineal"
|
||||
"ucSensTemperatura" = "Sensor de Temperatura"
|
||||
"ucBoton" = "Botón"
|
||||
|
||||
# Actuadores
|
||||
"ucVMmotorSim" = "Motor VetroMeccanica"
|
||||
"ucValvulaFluido" = "Válvula de Fluido"
|
||||
|
||||
# Elementos Estáticos
|
||||
"ucGuia" = "Guía"
|
||||
"ucDescarte" = "Descarte"
|
||||
|
||||
# Emuladores
|
||||
"ucBottGenerator" = "Generador de Botellas"
|
||||
"ucFiller" = "Llenadora"
|
||||
"ucTanque" = "Tanque"
|
||||
|
||||
# Señales
|
||||
"ucAnalogTag" = "Tag Analógico"
|
||||
"ucBoolTag" = "Tag Digital"
|
||||
"ucConsensGeneric" = "Consenso Genérico"
|
||||
|
||||
# Datos
|
||||
"ucExtraccionTag" = "Extracción de Tag"
|
||||
"ucBuscarCoincidencias" = "Búsqueda de Coincidencias"
|
||||
|
||||
# Decorativos
|
||||
"ucCustomImage" = "Imagen Personalizada"
|
||||
"ucFramePlate" = "Marco"
|
||||
"ucTextPlate" = "Placa de Texto"
|
||||
|
||||
# Fluidos
|
||||
"ucTuberiaFluido" = "Tubería de Fluido"
|
||||
"osSistemaFluidos" = "Sistema de Fluidos"
|
||||
|
||||
# Trazas
|
||||
"ucTrace3" = "Traza 3 Puntos"
|
||||
"ucTraceSimple" = "Traza Simple"
|
||||
|
||||
# Dinámicos
|
||||
"ucBotella" = "Botella"
|
||||
"ucBotellaCuello" = "Botella con Cuello"
|
||||
|
||||
# Ejemplo
|
||||
"ucBasicExample" = "Ejemplo Básico"
|
||||
}
|
||||
|
||||
# Mapeo de categorías estándar
|
||||
$categoriasEstandar = @{
|
||||
"Id:" = "Identificación"
|
||||
"Layout:" = "Posición y Tamaño"
|
||||
"Setup:" = "Configuración"
|
||||
"Simulation:" = "Simulación"
|
||||
"PLC link:" = "Enlace PLC"
|
||||
"Debug:" = "Información"
|
||||
"Encoder:" = "Encoder"
|
||||
"General:" = "Identificación"
|
||||
}
|
||||
|
||||
function Actualizar-NombreClase {
|
||||
param(
|
||||
[string]$archivo,
|
||||
[string]$nuevoNombre
|
||||
)
|
||||
|
||||
$contenido = Get-Content $archivo -Raw
|
||||
$patron = 'return\s+"[^"]*";'
|
||||
$reemplazo = "return `"$nuevoNombre`";"
|
||||
|
||||
if ($contenido -match $patron) {
|
||||
$contenido = $contenido -replace $patron, $reemplazo
|
||||
Set-Content $archivo $contenido -NoNewline
|
||||
Write-Host "✓ Actualizado NombreClase en $archivo" -ForegroundColor Yellow
|
||||
}
|
||||
}
|
||||
|
||||
function Estandarizar-Categorias {
|
||||
param([string]$archivo)
|
||||
|
||||
$contenido = Get-Content $archivo -Raw
|
||||
$modificado = $false
|
||||
|
||||
foreach ($categoria in $categoriasEstandar.Keys) {
|
||||
$patron = [regex]::Escape("Category(`"$categoria`")")
|
||||
$reemplazo = "Category(`"$($categoriasEstandar[$categoria])`")"
|
||||
|
||||
if ($contenido -match $patron) {
|
||||
$contenido = $contenido -replace $patron, $reemplazo
|
||||
$modificado = $true
|
||||
}
|
||||
}
|
||||
|
||||
if ($modificado) {
|
||||
Set-Content $archivo $contenido -NoNewline
|
||||
Write-Host "✓ Estandarizadas categorías en $archivo" -ForegroundColor Cyan
|
||||
}
|
||||
}
|
||||
|
||||
# Buscar todos los archivos .cs en ObjetosSim
|
||||
$archivos = Get-ChildItem -Path "ObjetosSim" -Filter "*.cs" -Recurse | Where-Object {
|
||||
$_.Name -like "uc*.cs" -or $_.Name -like "os*.cs"
|
||||
}
|
||||
|
||||
Write-Host "Encontrados $($archivos.Count) archivos para procesar" -ForegroundColor Blue
|
||||
|
||||
foreach ($archivo in $archivos) {
|
||||
$nombreArchivo = [System.IO.Path]::GetFileNameWithoutExtension($archivo.Name)
|
||||
|
||||
# Actualizar nombre de clase si está en el mapeo
|
||||
if ($nombresClase.ContainsKey($nombreArchivo)) {
|
||||
Actualizar-NombreClase -archivo $archivo.FullName -nuevoNombre $nombresClase[$nombreArchivo]
|
||||
}
|
||||
|
||||
# Estandarizar categorías
|
||||
Estandarizar-Categorias -archivo $archivo.FullName
|
||||
}
|
||||
|
||||
Write-Host "=== Proceso completado ===" -ForegroundColor Green
|
||||
Write-Host "Revise los cambios y verifique que la compilación sea exitosa" -ForegroundColor Yellow
|
||||
|
|
@ -1,629 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Sistema de Pruebas Hidráulicas para CtrEditor
|
||||
Basado en FluidManagementSystem.md y MCP_LLM_Guide.md
|
||||
|
||||
Este script ejecuta pruebas sistemáticas del sistema hidráulico:
|
||||
- Equilibrio de flujo entre tanques
|
||||
- Cálculos de unidades correctos
|
||||
- Comportamiento con diferentes tipos de fluidos
|
||||
- Verificación de niveles después de tiempo determinado
|
||||
"""
|
||||
|
||||
import json
|
||||
import time
|
||||
import requests
|
||||
import math
|
||||
from typing import Dict, List, Tuple, Any
|
||||
from dataclasses import dataclass, asdict
|
||||
|
||||
|
||||
@dataclass
|
||||
class FluidProperties:
|
||||
"""Propiedades de un fluido"""
|
||||
|
||||
name: str
|
||||
density: float # kg/m³
|
||||
viscosity: float # Pa·s
|
||||
temperature: float # °C
|
||||
|
||||
|
||||
@dataclass
|
||||
class TankState:
|
||||
"""Estado de un tanque"""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
level_m: float
|
||||
volume_l: float
|
||||
max_volume_l: float
|
||||
fluid_primary: FluidProperties
|
||||
fluid_secondary: FluidProperties
|
||||
primary_percentage: float
|
||||
|
||||
|
||||
@dataclass
|
||||
class PumpState:
|
||||
"""Estado de una bomba"""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
is_running: bool
|
||||
current_flow: float # m³/s
|
||||
max_flow: float # m³/s
|
||||
pump_head: float # m
|
||||
|
||||
|
||||
@dataclass
|
||||
class PipeState:
|
||||
"""Estado de una tubería"""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
current_flow: float # m³/s
|
||||
pressure_drop: float # Pa
|
||||
fluid_type: str
|
||||
|
||||
|
||||
class HydraulicTestManager:
|
||||
"""Administrador de pruebas hidráulicas"""
|
||||
|
||||
def __init__(self, mcp_url: str = "http://localhost:3000"):
|
||||
self.mcp_url = mcp_url
|
||||
self.test_results = []
|
||||
|
||||
def send_mcp_request(self, method: str, params: Dict = None) -> Dict:
|
||||
"""Envía una solicitud MCP y retorna la respuesta"""
|
||||
payload = {"jsonrpc": "2.0", "id": 1, "method": method, "params": params or {}}
|
||||
|
||||
try:
|
||||
response = requests.post(self.mcp_url, json=payload, timeout=10)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except Exception as e:
|
||||
print(f"Error en solicitud MCP: {e}")
|
||||
return {"error": str(e)}
|
||||
|
||||
def get_simulation_objects(self) -> List[Dict]:
|
||||
"""Obtiene todos los objetos de la simulación"""
|
||||
response = self.send_mcp_request("list_objects")
|
||||
if "result" in response:
|
||||
return response["result"].get("objects", [])
|
||||
return []
|
||||
|
||||
def get_tanks(self) -> List[TankState]:
|
||||
"""Obtiene el estado de todos los tanques"""
|
||||
objects = self.get_simulation_objects()
|
||||
tanks = []
|
||||
|
||||
for obj in objects:
|
||||
if obj.get("type") == "osHydTank":
|
||||
fluid_props = obj.get("properties", {})
|
||||
|
||||
# Crear propiedades de fluido primario
|
||||
primary_fluid = FluidProperties(
|
||||
name=fluid_props.get("PrimaryFluidName", "Water"),
|
||||
density=fluid_props.get("PrimaryFluidDensity", 1000.0),
|
||||
viscosity=fluid_props.get("PrimaryFluidViscosity", 0.001),
|
||||
temperature=fluid_props.get("PrimaryFluidTemperature", 20.0),
|
||||
)
|
||||
|
||||
# Crear propiedades de fluido secundario
|
||||
secondary_fluid = FluidProperties(
|
||||
name=fluid_props.get("SecondaryFluidName", "Air"),
|
||||
density=fluid_props.get("SecondaryFluidDensity", 1.225),
|
||||
viscosity=fluid_props.get("SecondaryFluidViscosity", 1.8e-5),
|
||||
temperature=fluid_props.get("SecondaryFluidTemperature", 20.0),
|
||||
)
|
||||
|
||||
tank = TankState(
|
||||
id=obj["id"],
|
||||
name=obj["name"],
|
||||
level_m=fluid_props.get("CurrentLevelM", 0.0),
|
||||
volume_l=fluid_props.get("CurrentVolumeL", 0.0),
|
||||
max_volume_l=fluid_props.get("MaxVolumeL", 1000.0),
|
||||
fluid_primary=primary_fluid,
|
||||
fluid_secondary=secondary_fluid,
|
||||
primary_percentage=fluid_props.get("PrimaryFluidPercentage", 100.0),
|
||||
)
|
||||
tanks.append(tank)
|
||||
|
||||
return tanks
|
||||
|
||||
def get_pumps(self) -> List[PumpState]:
|
||||
"""Obtiene el estado de todas las bombas"""
|
||||
objects = self.get_simulation_objects()
|
||||
pumps = []
|
||||
|
||||
for obj in objects:
|
||||
if obj.get("type") == "osHydPump":
|
||||
props = obj.get("properties", {})
|
||||
pump = PumpState(
|
||||
id=obj["id"],
|
||||
name=obj["name"],
|
||||
is_running=props.get("IsRunning", False),
|
||||
current_flow=props.get("CurrentFlow", 0.0),
|
||||
max_flow=props.get("MaxFlow", 0.02),
|
||||
pump_head=props.get("PumpHead", 75.0),
|
||||
)
|
||||
pumps.append(pump)
|
||||
|
||||
return pumps
|
||||
|
||||
def get_pipes(self) -> List[PipeState]:
|
||||
"""Obtiene el estado de todas las tuberías"""
|
||||
objects = self.get_simulation_objects()
|
||||
pipes = []
|
||||
|
||||
for obj in objects:
|
||||
if obj.get("type") == "osHydPipe":
|
||||
props = obj.get("properties", {})
|
||||
pipe = PipeState(
|
||||
id=obj["id"],
|
||||
name=obj["name"],
|
||||
current_flow=props.get("CurrentFlow", 0.0),
|
||||
pressure_drop=props.get("PressureDrop", 0.0),
|
||||
fluid_type=props.get("FluidType", "Unknown"),
|
||||
)
|
||||
pipes.append(pipe)
|
||||
|
||||
return pipes
|
||||
|
||||
def start_simulation(self) -> bool:
|
||||
"""Inicia la simulación"""
|
||||
response = self.send_mcp_request("start_simulation")
|
||||
return "result" in response and response["result"].get("success", False)
|
||||
|
||||
def stop_simulation(self) -> bool:
|
||||
"""Detiene la simulación"""
|
||||
response = self.send_mcp_request("stop_simulation")
|
||||
return "result" in response and response["result"].get("success", False)
|
||||
|
||||
def reset_simulation_timing(self) -> bool:
|
||||
"""Resetea los contadores de tiempo de simulación"""
|
||||
response = self.send_mcp_request("reset_simulation_timing")
|
||||
return "result" in response and response["result"].get("success", False)
|
||||
|
||||
def wait_simulation_time(self, target_seconds: float):
|
||||
"""Espera hasta que la simulación haya ejecutado un tiempo determinado"""
|
||||
print(f"⏰ Esperando {target_seconds} segundos reales de simulación...")
|
||||
|
||||
target_ms = target_seconds * 1000
|
||||
start_time = time.time()
|
||||
max_wait_time = target_seconds * 3 # Timeout si no progresa
|
||||
|
||||
while True:
|
||||
# Verificar tiempo real transcurrido para timeout
|
||||
if time.time() - start_time > max_wait_time:
|
||||
print(f"⚠️ Timeout después de {max_wait_time} segundos reales")
|
||||
break
|
||||
|
||||
# Obtener estado actual de simulación
|
||||
response = self.send_mcp_request("get_simulation_status")
|
||||
if "result" in response:
|
||||
result = response["result"]
|
||||
simulation_ms = result.get("simulation_elapsed_ms", 0)
|
||||
is_running = result.get("is_running", False)
|
||||
|
||||
if simulation_ms >= target_ms:
|
||||
print(
|
||||
f"✅ Simulación completó {simulation_ms}ms ({simulation_ms/1000:.3f}s)"
|
||||
)
|
||||
break
|
||||
|
||||
if not is_running:
|
||||
print(f"⚠️ Simulación detenida en {simulation_ms}ms")
|
||||
break
|
||||
|
||||
# Esperar un poco antes de verificar nuevamente
|
||||
time.sleep(0.1)
|
||||
else:
|
||||
print("⚠️ Error obteniendo estado de simulación")
|
||||
time.sleep(0.5)
|
||||
|
||||
def calculate_mass_balance(
|
||||
self, initial_tanks: List[TankState], final_tanks: List[TankState]
|
||||
) -> Dict:
|
||||
"""Calcula el balance de masa entre estados inicial y final"""
|
||||
initial_mass = 0.0
|
||||
final_mass = 0.0
|
||||
|
||||
for i, tank in enumerate(initial_tanks):
|
||||
# Masa inicial = volumen × densidad del fluido primario × porcentaje
|
||||
tank_mass = (
|
||||
(tank.volume_l / 1000.0)
|
||||
* tank.fluid_primary.density
|
||||
* (tank.primary_percentage / 100.0)
|
||||
)
|
||||
initial_mass += tank_mass
|
||||
|
||||
for i, tank in enumerate(final_tanks):
|
||||
# Masa final = volumen × densidad del fluido primario × porcentaje
|
||||
tank_mass = (
|
||||
(tank.volume_l / 1000.0)
|
||||
* tank.fluid_primary.density
|
||||
* (tank.primary_percentage / 100.0)
|
||||
)
|
||||
final_mass += tank_mass
|
||||
|
||||
return {
|
||||
"initial_mass_kg": initial_mass,
|
||||
"final_mass_kg": final_mass,
|
||||
"mass_difference_kg": final_mass - initial_mass,
|
||||
"conservation_percentage": (
|
||||
(final_mass / initial_mass * 100.0) if initial_mass > 0 else 0.0
|
||||
),
|
||||
}
|
||||
|
||||
def test_basic_flow_equilibrium(self) -> Dict:
|
||||
"""
|
||||
Prueba básica de equilibrio de flujo entre dos tanques
|
||||
"""
|
||||
print("\n🧪 EJECUTANDO: Prueba de Equilibrio de Flujo Básico")
|
||||
|
||||
test_result = {
|
||||
"test_name": "basic_flow_equilibrium",
|
||||
"description": "Verificar equilibrio de flujo entre dos tanques con bomba",
|
||||
"success": False,
|
||||
"details": {},
|
||||
"measurements": {},
|
||||
}
|
||||
|
||||
try:
|
||||
# Resetear timing para medición precisa
|
||||
self.reset_simulation_timing()
|
||||
|
||||
# Estado inicial
|
||||
initial_tanks = self.get_tanks()
|
||||
initial_pumps = self.get_pumps()
|
||||
initial_pipes = self.get_pipes()
|
||||
|
||||
if len(initial_tanks) < 2:
|
||||
test_result["details"][
|
||||
"error"
|
||||
] = "Se requieren al menos 2 tanques para la prueba"
|
||||
return test_result
|
||||
|
||||
print(f"📊 Estado inicial:")
|
||||
for tank in initial_tanks:
|
||||
print(f" - {tank.name}: {tank.level_m:.2f}m ({tank.volume_l:.1f}L)")
|
||||
|
||||
# Iniciar simulación
|
||||
if not self.start_simulation():
|
||||
test_result["details"]["error"] = "No se pudo iniciar la simulación"
|
||||
return test_result
|
||||
|
||||
# Esperar tiempo real de simulación
|
||||
target_simulation_time = 30.0 # 30 segundos reales de simulación
|
||||
self.wait_simulation_time(target_simulation_time)
|
||||
|
||||
# Estado final
|
||||
final_tanks = self.get_tanks()
|
||||
final_pumps = self.get_pumps()
|
||||
final_pipes = self.get_pipes()
|
||||
|
||||
print(f"📊 Estado final:")
|
||||
for tank in final_tanks:
|
||||
print(f" - {tank.name}: {tank.level_m:.2f}m ({tank.volume_l:.1f}L)")
|
||||
|
||||
# Detener simulación
|
||||
self.stop_simulation()
|
||||
|
||||
# Calcular balance de masa
|
||||
mass_balance = self.calculate_mass_balance(initial_tanks, final_tanks)
|
||||
|
||||
# Obtener tiempo real de simulación
|
||||
status_response = self.send_mcp_request("get_simulation_status")
|
||||
actual_simulation_ms = 0
|
||||
if "result" in status_response:
|
||||
actual_simulation_ms = status_response["result"].get(
|
||||
"simulation_elapsed_ms", 0
|
||||
)
|
||||
|
||||
# Almacenar mediciones
|
||||
test_result["measurements"] = {
|
||||
"target_simulation_time_s": target_simulation_time,
|
||||
"actual_simulation_time_ms": actual_simulation_ms,
|
||||
"actual_simulation_time_s": actual_simulation_ms / 1000.0,
|
||||
"initial_tanks": [asdict(tank) for tank in initial_tanks],
|
||||
"final_tanks": [asdict(tank) for tank in final_tanks],
|
||||
"pumps": [asdict(pump) for pump in final_pumps],
|
||||
"pipes": [asdict(pipe) for pipe in final_pipes],
|
||||
"mass_balance": mass_balance,
|
||||
}
|
||||
|
||||
# Verificar conservación de masa (tolerancia 5%)
|
||||
conservation_ok = (
|
||||
abs(mass_balance["conservation_percentage"] - 100.0) <= 5.0
|
||||
)
|
||||
|
||||
# Verificar que hubo transferencia de fluido
|
||||
volume_change = any(
|
||||
abs(final.volume_l - initial.volume_l) > 1.0
|
||||
for initial, final in zip(initial_tanks, final_tanks)
|
||||
)
|
||||
|
||||
test_result["success"] = conservation_ok and volume_change
|
||||
test_result["details"] = {
|
||||
"conservation_percentage": mass_balance["conservation_percentage"],
|
||||
"volume_transfer_detected": volume_change,
|
||||
"mass_conserved": conservation_ok,
|
||||
}
|
||||
|
||||
if test_result["success"]:
|
||||
print(
|
||||
"✅ Prueba EXITOSA: Equilibrio de flujo funcionando correctamente"
|
||||
)
|
||||
else:
|
||||
print("❌ Prueba FALLIDA: Problemas en equilibrio de flujo")
|
||||
|
||||
except Exception as e:
|
||||
test_result["details"]["error"] = str(e)
|
||||
print(f"❌ Error en prueba: {e}")
|
||||
|
||||
return test_result
|
||||
|
||||
def test_fluid_properties_consistency(self) -> Dict:
|
||||
"""
|
||||
Prueba de consistencia de propiedades de fluidos
|
||||
"""
|
||||
print("\n🧪 EJECUTANDO: Prueba de Consistencia de Propiedades de Fluidos")
|
||||
|
||||
test_result = {
|
||||
"test_name": "fluid_properties_consistency",
|
||||
"description": "Verificar que las propiedades de fluidos se mantienen consistentes",
|
||||
"success": False,
|
||||
"details": {},
|
||||
"measurements": {},
|
||||
}
|
||||
|
||||
try:
|
||||
# Obtener estado actual
|
||||
tanks = self.get_tanks()
|
||||
pipes = self.get_pipes()
|
||||
|
||||
if len(tanks) < 2:
|
||||
test_result["details"][
|
||||
"error"
|
||||
] = "Se requieren al menos 2 tanques para la prueba"
|
||||
return test_result
|
||||
|
||||
# Verificar propiedades de fluidos en tanques
|
||||
fluid_consistency = True
|
||||
density_checks = []
|
||||
|
||||
for tank in tanks:
|
||||
# Verificar que las densidades sean razonables
|
||||
primary_density_ok = (
|
||||
500.0 <= tank.fluid_primary.density <= 2000.0
|
||||
) # kg/m³
|
||||
secondary_density_ok = (
|
||||
0.5 <= tank.fluid_secondary.density <= 2000.0
|
||||
) # kg/m³
|
||||
|
||||
density_checks.append(
|
||||
{
|
||||
"tank_name": tank.name,
|
||||
"primary_density": tank.fluid_primary.density,
|
||||
"secondary_density": tank.fluid_secondary.density,
|
||||
"primary_ok": primary_density_ok,
|
||||
"secondary_ok": secondary_density_ok,
|
||||
}
|
||||
)
|
||||
|
||||
if not (primary_density_ok and secondary_density_ok):
|
||||
fluid_consistency = False
|
||||
|
||||
# Verificar que las tuberías muestren información de fluido
|
||||
pipe_fluid_info = []
|
||||
for pipe in pipes:
|
||||
has_fluid_info = pipe.fluid_type != "Unknown" and pipe.fluid_type != ""
|
||||
pipe_fluid_info.append(
|
||||
{
|
||||
"pipe_name": pipe.name,
|
||||
"fluid_type": pipe.fluid_type,
|
||||
"has_fluid_info": has_fluid_info,
|
||||
}
|
||||
)
|
||||
|
||||
test_result["measurements"] = {
|
||||
"density_checks": density_checks,
|
||||
"pipe_fluid_info": pipe_fluid_info,
|
||||
}
|
||||
|
||||
pipe_info_ok = (
|
||||
all(info["has_fluid_info"] for info in pipe_fluid_info)
|
||||
if pipe_fluid_info
|
||||
else True
|
||||
)
|
||||
|
||||
test_result["success"] = fluid_consistency and pipe_info_ok
|
||||
test_result["details"] = {
|
||||
"fluid_densities_valid": fluid_consistency,
|
||||
"pipe_fluid_info_available": pipe_info_ok,
|
||||
"tanks_checked": len(tanks),
|
||||
"pipes_checked": len(pipes),
|
||||
}
|
||||
|
||||
if test_result["success"]:
|
||||
print("✅ Prueba EXITOSA: Propiedades de fluidos consistentes")
|
||||
else:
|
||||
print("❌ Prueba FALLIDA: Inconsistencias en propiedades de fluidos")
|
||||
|
||||
except Exception as e:
|
||||
test_result["details"]["error"] = str(e)
|
||||
print(f"❌ Error en prueba: {e}")
|
||||
|
||||
return test_result
|
||||
|
||||
def test_mixed_fluid_behavior(self) -> Dict:
|
||||
"""
|
||||
Prueba de comportamiento con fluidos mezclados
|
||||
"""
|
||||
print("\n🧪 EJECUTANDO: Prueba de Comportamiento de Fluidos Mezclados")
|
||||
|
||||
test_result = {
|
||||
"test_name": "mixed_fluid_behavior",
|
||||
"description": "Verificar comportamiento con fluidos primarios y secundarios",
|
||||
"success": False,
|
||||
"details": {},
|
||||
"measurements": {},
|
||||
}
|
||||
|
||||
try:
|
||||
# Obtener tanques
|
||||
tanks = self.get_tanks()
|
||||
|
||||
if len(tanks) < 2:
|
||||
test_result["details"][
|
||||
"error"
|
||||
] = "Se requieren al menos 2 tanques para la prueba"
|
||||
return test_result
|
||||
|
||||
# Verificar que tenemos fluidos mezclados
|
||||
mixed_tanks = []
|
||||
for tank in tanks:
|
||||
has_mixed_fluid = (
|
||||
tank.primary_percentage > 0.0 and tank.primary_percentage < 100.0
|
||||
)
|
||||
|
||||
mixed_tanks.append(
|
||||
{
|
||||
"tank_name": tank.name,
|
||||
"primary_percentage": tank.primary_percentage,
|
||||
"has_mixed_fluid": has_mixed_fluid,
|
||||
"primary_fluid": tank.fluid_primary.name,
|
||||
"secondary_fluid": tank.fluid_secondary.name,
|
||||
}
|
||||
)
|
||||
|
||||
# Ejecutar simulación corta
|
||||
if self.start_simulation():
|
||||
self.wait_simulation_time(10.0)
|
||||
|
||||
# Verificar que los porcentajes se mantienen en rangos válidos
|
||||
final_tanks = self.get_tanks()
|
||||
percentage_stability = []
|
||||
|
||||
for initial, final in zip(tanks, final_tanks):
|
||||
percentage_change = abs(
|
||||
final.primary_percentage - initial.primary_percentage
|
||||
)
|
||||
stable = percentage_change <= 10.0 # Tolerancia 10%
|
||||
|
||||
percentage_stability.append(
|
||||
{
|
||||
"tank_name": initial.name,
|
||||
"initial_percentage": initial.primary_percentage,
|
||||
"final_percentage": final.primary_percentage,
|
||||
"change": percentage_change,
|
||||
"stable": stable,
|
||||
}
|
||||
)
|
||||
|
||||
self.stop_simulation()
|
||||
|
||||
test_result["measurements"] = {
|
||||
"mixed_tanks": mixed_tanks,
|
||||
"percentage_stability": percentage_stability,
|
||||
}
|
||||
|
||||
# Determinar éxito
|
||||
has_mixed_fluids = any(tank["has_mixed_fluid"] for tank in mixed_tanks)
|
||||
percentages_stable = all(
|
||||
check["stable"] for check in percentage_stability
|
||||
)
|
||||
|
||||
test_result["success"] = has_mixed_fluids or percentages_stable
|
||||
test_result["details"] = {
|
||||
"mixed_fluids_detected": has_mixed_fluids,
|
||||
"percentages_stable": percentages_stable,
|
||||
"tanks_with_mixed_fluids": sum(
|
||||
1 for tank in mixed_tanks if tank["has_mixed_fluid"]
|
||||
),
|
||||
}
|
||||
|
||||
if test_result["success"]:
|
||||
print(
|
||||
"✅ Prueba EXITOSA: Comportamiento de fluidos mezclados correcto"
|
||||
)
|
||||
else:
|
||||
print("❌ Prueba FALLIDA: Problemas con fluidos mezclados")
|
||||
else:
|
||||
test_result["details"]["error"] = "No se pudo iniciar la simulación"
|
||||
|
||||
except Exception as e:
|
||||
test_result["details"]["error"] = str(e)
|
||||
print(f"❌ Error en prueba: {e}")
|
||||
|
||||
return test_result
|
||||
|
||||
def run_all_tests(self) -> Dict:
|
||||
"""Ejecuta todas las pruebas del sistema hidráulico"""
|
||||
print("🚀 INICIANDO PRUEBAS DEL SISTEMA HIDRÁULICO")
|
||||
print("=" * 60)
|
||||
|
||||
all_results = {
|
||||
"test_suite": "HydraulicSystemTests",
|
||||
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"tests": [],
|
||||
"summary": {},
|
||||
}
|
||||
|
||||
# Lista de pruebas a ejecutar
|
||||
tests = [
|
||||
self.test_basic_flow_equilibrium,
|
||||
self.test_fluid_properties_consistency,
|
||||
self.test_mixed_fluid_behavior,
|
||||
]
|
||||
|
||||
successful_tests = 0
|
||||
|
||||
for test_func in tests:
|
||||
result = test_func()
|
||||
all_results["tests"].append(result)
|
||||
|
||||
if result["success"]:
|
||||
successful_tests += 1
|
||||
|
||||
# Resumen
|
||||
all_results["summary"] = {
|
||||
"total_tests": len(tests),
|
||||
"successful_tests": successful_tests,
|
||||
"failed_tests": len(tests) - successful_tests,
|
||||
"success_rate": (successful_tests / len(tests)) * 100.0,
|
||||
}
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print("📋 RESUMEN DE PRUEBAS")
|
||||
print(f"Total de pruebas: {all_results['summary']['total_tests']}")
|
||||
print(f"Exitosas: {all_results['summary']['successful_tests']}")
|
||||
print(f"Fallidas: {all_results['summary']['failed_tests']}")
|
||||
print(f"Tasa de éxito: {all_results['summary']['success_rate']:.1f}%")
|
||||
|
||||
return all_results
|
||||
|
||||
|
||||
def main():
|
||||
"""Función principal"""
|
||||
print("🔧 Sistema de Pruebas Hidráulicas para CtrEditor")
|
||||
print("=" * 50)
|
||||
|
||||
# Crear administrador de pruebas
|
||||
test_manager = HydraulicTestManager()
|
||||
|
||||
# Ejecutar todas las pruebas
|
||||
results = test_manager.run_all_tests()
|
||||
|
||||
# Guardar resultados
|
||||
results_file = f"hydraulic_test_results_{int(time.time())}.json"
|
||||
with open(results_file, "w", encoding="utf-8") as f:
|
||||
json.dump(results, f, indent=2, ensure_ascii=False)
|
||||
|
||||
print(f"\n💾 Resultados guardados en: {results_file}")
|
||||
|
||||
return results
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -1,425 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Sistema de Pruebas Hidráulicas con Timing Preciso
|
||||
================================================
|
||||
|
||||
Utiliza el tiempo real de simulación reportado por CtrEditor
|
||||
para mediciones precisas e independientes del sistema.
|
||||
|
||||
Autor: Sistema de Automatización CtrEditor
|
||||
Fecha: Enero 2025
|
||||
"""
|
||||
|
||||
import json
|
||||
import socket
|
||||
import time
|
||||
from dataclasses import dataclass, asdict
|
||||
from typing import List, Dict, Any, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class TankInfo:
|
||||
name: str
|
||||
level_m: float
|
||||
volume_l: float
|
||||
primary_fluid: str
|
||||
secondary_fluid: str
|
||||
primary_volume: float
|
||||
secondary_volume: float
|
||||
mixing_volume: float
|
||||
|
||||
|
||||
@dataclass
|
||||
class PumpInfo:
|
||||
name: str
|
||||
is_running: bool
|
||||
current_flow: float
|
||||
max_flow: float
|
||||
pump_head: float
|
||||
|
||||
|
||||
@dataclass
|
||||
class PipeInfo:
|
||||
name: str
|
||||
current_flow: float
|
||||
pressure_drop: float
|
||||
|
||||
|
||||
class HydraulicTestSystemV2:
|
||||
def __init__(self, host="localhost", port=5005):
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.test_results = []
|
||||
|
||||
def send_mcp_request(self, method: str, params: Dict = None) -> Dict:
|
||||
"""Envía una solicitud MCP al servidor proxy"""
|
||||
if params is None:
|
||||
params = {"random_string": "test"}
|
||||
|
||||
request = {
|
||||
"jsonrpc": "2.0",
|
||||
"id": 1,
|
||||
"method": "tools/call",
|
||||
"params": {"name": method, "arguments": params},
|
||||
}
|
||||
|
||||
try:
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||||
sock.settimeout(10)
|
||||
sock.connect((self.host, self.port))
|
||||
|
||||
request_str = json.dumps(request) + "\n"
|
||||
sock.sendall(request_str.encode())
|
||||
|
||||
response = sock.recv(4096).decode()
|
||||
return json.loads(response)
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error en comunicación MCP: {e}")
|
||||
return {"error": str(e)}
|
||||
|
||||
def get_simulation_timing(self) -> Dict[str, Any]:
|
||||
"""Obtiene información detallada del timing de simulación"""
|
||||
response = self.send_mcp_request("get_simulation_status")
|
||||
if "result" in response:
|
||||
result = response["result"]
|
||||
return {
|
||||
"is_running": result.get("is_running", False),
|
||||
"elapsed_ms": result.get("simulation_elapsed_ms", 0),
|
||||
"elapsed_seconds": result.get("simulation_elapsed_seconds", 0.0),
|
||||
}
|
||||
return {"is_running": False, "elapsed_ms": 0, "elapsed_seconds": 0.0}
|
||||
|
||||
def wait_simulation_time(
|
||||
self, target_seconds: float, max_real_time: float = None
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Espera hasta que la simulación haya ejecutado un tiempo determinado
|
||||
|
||||
Args:
|
||||
target_seconds: Tiempo objetivo de simulación en segundos
|
||||
max_real_time: Tiempo máximo real a esperar (default: target_seconds * 5)
|
||||
|
||||
Returns:
|
||||
Dict con información del timing final
|
||||
"""
|
||||
if max_real_time is None:
|
||||
max_real_time = target_seconds * 5 # 5x timeout por defecto
|
||||
|
||||
print(f"⏰ Esperando {target_seconds}s de simulación real...")
|
||||
|
||||
target_ms = target_seconds * 1000
|
||||
start_real_time = time.time()
|
||||
last_progress_time = start_real_time
|
||||
last_simulation_ms = 0
|
||||
|
||||
while True:
|
||||
current_real_time = time.time()
|
||||
|
||||
# Timeout si excede tiempo máximo real
|
||||
if current_real_time - start_real_time > max_real_time:
|
||||
print(f"⚠️ Timeout después de {max_real_time:.1f}s reales")
|
||||
break
|
||||
|
||||
# Obtener estado actual
|
||||
timing = self.get_simulation_timing()
|
||||
current_simulation_ms = timing["elapsed_ms"]
|
||||
|
||||
# Verificar si la simulación sigue corriendo
|
||||
if not timing["is_running"]:
|
||||
print(f"⚠️ Simulación detenida en {current_simulation_ms}ms")
|
||||
break
|
||||
|
||||
# Verificar progreso (evitar simulaciones colgadas)
|
||||
if current_simulation_ms > last_simulation_ms:
|
||||
last_progress_time = current_real_time
|
||||
last_simulation_ms = current_simulation_ms
|
||||
elif current_real_time - last_progress_time > 5.0: # 5s sin progreso
|
||||
print(f"⚠️ Sin progreso en simulación por 5 segundos")
|
||||
break
|
||||
|
||||
# Verificar si alcanzamos el objetivo
|
||||
if current_simulation_ms >= target_ms:
|
||||
print(
|
||||
f"✅ Objetivo alcanzado: {current_simulation_ms}ms ({timing['elapsed_seconds']:.3f}s)"
|
||||
)
|
||||
break
|
||||
|
||||
# Mostrar progreso cada segundo
|
||||
if (
|
||||
int(current_real_time) > int(start_real_time)
|
||||
and int(current_real_time) % 2 == 0
|
||||
):
|
||||
progress = (current_simulation_ms / target_ms) * 100
|
||||
print(
|
||||
f"📊 Progreso: {progress:.1f}% ({current_simulation_ms}ms de {target_ms}ms)"
|
||||
)
|
||||
|
||||
time.sleep(0.1) # Check cada 100ms
|
||||
|
||||
return timing
|
||||
|
||||
def get_tanks(self) -> List[TankInfo]:
|
||||
"""Obtiene información de todos los tanques"""
|
||||
response = self.send_mcp_request("list_objects")
|
||||
tanks = []
|
||||
|
||||
if "result" in response and "objects" in response["result"]:
|
||||
for obj in response["result"]["objects"]:
|
||||
if obj["type"] == "osHydTank":
|
||||
props = obj["properties"]
|
||||
tank = TankInfo(
|
||||
name=props.get("name", "Unknown"),
|
||||
level_m=props.get("CurrentLevelM", 0.0),
|
||||
volume_l=props.get("TotalVolumeL", 0.0),
|
||||
primary_fluid=props.get("PrimaryFluidType", "Unknown"),
|
||||
secondary_fluid=props.get("SecondaryFluidType", "Unknown"),
|
||||
primary_volume=props.get("PrimaryVolumeL", 0.0),
|
||||
secondary_volume=props.get("SecondaryVolumeL", 0.0),
|
||||
mixing_volume=props.get("MixingVolumeL", 0.0),
|
||||
)
|
||||
tanks.append(tank)
|
||||
|
||||
return tanks
|
||||
|
||||
def get_pumps(self) -> List[PumpInfo]:
|
||||
"""Obtiene información de todas las bombas"""
|
||||
response = self.send_mcp_request("list_objects")
|
||||
pumps = []
|
||||
|
||||
if "result" in response and "objects" in response["result"]:
|
||||
for obj in response["result"]["objects"]:
|
||||
if obj["type"] == "osHydPump":
|
||||
props = obj["properties"]
|
||||
pump = PumpInfo(
|
||||
name=props.get("name", "Unknown"),
|
||||
is_running=props.get("IsRunning", False),
|
||||
current_flow=props.get("CurrentFlow", 0.0),
|
||||
max_flow=props.get("MaxFlow", 0.0),
|
||||
pump_head=props.get("PumpHead", 0.0),
|
||||
)
|
||||
pumps.append(pump)
|
||||
|
||||
return pumps
|
||||
|
||||
def get_pipes(self) -> List[PipeInfo]:
|
||||
"""Obtiene información de todas las tuberías"""
|
||||
response = self.send_mcp_request("list_objects")
|
||||
pipes = []
|
||||
|
||||
if "result" in response and "objects" in response["result"]:
|
||||
for obj in response["result"]["objects"]:
|
||||
if obj["type"] == "osHydPipe":
|
||||
props = obj["properties"]
|
||||
pipe = PipeInfo(
|
||||
name=props.get("name", "Unknown"),
|
||||
current_flow=props.get("CurrentFlow", 0.0),
|
||||
pressure_drop=props.get("PressureDrop", 0.0),
|
||||
)
|
||||
pipes.append(pipe)
|
||||
|
||||
return pipes
|
||||
|
||||
def start_simulation(self) -> bool:
|
||||
"""Inicia la simulación"""
|
||||
response = self.send_mcp_request("start_simulation")
|
||||
return "result" in response and response["result"].get("success", False)
|
||||
|
||||
def stop_simulation(self) -> Dict[str, Any]:
|
||||
"""Detiene la simulación y retorna timing final"""
|
||||
response = self.send_mcp_request("stop_simulation")
|
||||
if "result" in response and response["result"].get("success", False):
|
||||
return {
|
||||
"success": True,
|
||||
"elapsed_ms": response["result"].get("simulation_elapsed_ms", 0),
|
||||
"elapsed_seconds": response["result"].get(
|
||||
"simulation_elapsed_seconds", 0.0
|
||||
),
|
||||
}
|
||||
return {"success": False, "elapsed_ms": 0, "elapsed_seconds": 0.0}
|
||||
|
||||
def test_precise_flow_equilibrium(
|
||||
self, target_simulation_seconds: float = 30.0
|
||||
) -> Dict:
|
||||
"""
|
||||
Prueba de equilibrio de flujo con timing preciso
|
||||
|
||||
Args:
|
||||
target_simulation_seconds: Tiempo objetivo de simulación en segundos
|
||||
|
||||
Returns:
|
||||
Dict con resultados detallados de la prueba
|
||||
"""
|
||||
print(f"\n🧪 === PRUEBA DE EQUILIBRIO DE FLUJO (TIMING PRECISO) ===")
|
||||
print(f"⏱️ Tiempo objetivo: {target_simulation_seconds} segundos de simulación")
|
||||
|
||||
test_result = {
|
||||
"test_name": "Precise Flow Equilibrium",
|
||||
"success": False,
|
||||
"target_simulation_seconds": target_simulation_seconds,
|
||||
"actual_simulation_data": {},
|
||||
"measurements": {},
|
||||
"details": {},
|
||||
}
|
||||
|
||||
try:
|
||||
# Estado inicial
|
||||
initial_timing = self.get_simulation_timing()
|
||||
initial_tanks = self.get_tanks()
|
||||
initial_pumps = self.get_pumps()
|
||||
initial_pipes = self.get_pipes()
|
||||
|
||||
if len(initial_tanks) < 2:
|
||||
test_result["details"][
|
||||
"error"
|
||||
] = "Se requieren al menos 2 tanques para la prueba"
|
||||
return test_result
|
||||
|
||||
print(f"📊 Estado inicial:")
|
||||
print(f"⏱️ Tiempo simulación: {initial_timing['elapsed_seconds']:.3f}s")
|
||||
for tank in initial_tanks:
|
||||
print(f" - {tank.name}: {tank.level_m:.2f}m ({tank.volume_l:.1f}L)")
|
||||
|
||||
# Iniciar simulación
|
||||
if not self.start_simulation():
|
||||
test_result["details"]["error"] = "No se pudo iniciar la simulación"
|
||||
return test_result
|
||||
|
||||
print("🚀 Simulación iniciada...")
|
||||
|
||||
# Esperar tiempo de simulación preciso
|
||||
final_timing = self.wait_simulation_time(target_simulation_seconds)
|
||||
|
||||
# Detener simulación y obtener timing final
|
||||
stop_result = self.stop_simulation()
|
||||
|
||||
# Estado final
|
||||
final_tanks = self.get_tanks()
|
||||
final_pumps = self.get_pumps()
|
||||
final_pipes = self.get_pipes()
|
||||
|
||||
print(f"📊 Estado final:")
|
||||
print(f"⏱️ Tiempo total simulación: {stop_result['elapsed_seconds']:.3f}s")
|
||||
for tank in final_tanks:
|
||||
print(f" - {tank.name}: {tank.level_m:.2f}m ({tank.volume_l:.1f}L)")
|
||||
|
||||
# Calcular balance de masa
|
||||
initial_total_volume = sum(tank.volume_l for tank in initial_tanks)
|
||||
final_total_volume = sum(tank.volume_l for tank in final_tanks)
|
||||
mass_balance = {
|
||||
"initial_volume_l": initial_total_volume,
|
||||
"final_volume_l": final_total_volume,
|
||||
"difference_l": final_total_volume - initial_total_volume,
|
||||
"conservation_percentage": (
|
||||
(final_total_volume / initial_total_volume * 100)
|
||||
if initial_total_volume > 0
|
||||
else 0
|
||||
),
|
||||
}
|
||||
|
||||
print(f"⚖️ Balance de masa:")
|
||||
print(f" - Volumen inicial: {mass_balance['initial_volume_l']:.2f}L")
|
||||
print(f" - Volumen final: {mass_balance['final_volume_l']:.2f}L")
|
||||
print(f" - Diferencia: {mass_balance['difference_l']:.3f}L")
|
||||
print(f" - Conservación: {mass_balance['conservation_percentage']:.2f}%")
|
||||
|
||||
# Almacenar mediciones con timing preciso
|
||||
test_result["actual_simulation_data"] = {
|
||||
"initial_timing": initial_timing,
|
||||
"final_timing": final_timing,
|
||||
"stop_timing": stop_result,
|
||||
"actual_simulation_seconds": stop_result["elapsed_seconds"],
|
||||
"timing_accuracy": abs(
|
||||
target_simulation_seconds - stop_result["elapsed_seconds"]
|
||||
),
|
||||
}
|
||||
|
||||
test_result["measurements"] = {
|
||||
"target_simulation_seconds": target_simulation_seconds,
|
||||
"actual_simulation_seconds": stop_result["elapsed_seconds"],
|
||||
"initial_tanks": [asdict(tank) for tank in initial_tanks],
|
||||
"final_tanks": [asdict(tank) for tank in final_tanks],
|
||||
"pumps": [asdict(pump) for pump in final_pumps],
|
||||
"pipes": [asdict(pipe) for pipe in final_pipes],
|
||||
"mass_balance": mass_balance,
|
||||
}
|
||||
|
||||
# Verificar éxito (conservación de masa dentro del 1%)
|
||||
conservation_ok = (
|
||||
abs(mass_balance["difference_l"])
|
||||
< 0.01 * mass_balance["initial_volume_l"]
|
||||
)
|
||||
timing_ok = (
|
||||
abs(target_simulation_seconds - stop_result["elapsed_seconds"]) < 1.0
|
||||
) # Tolerancia de 1s
|
||||
|
||||
test_result["success"] = conservation_ok and timing_ok
|
||||
|
||||
if conservation_ok:
|
||||
print("✅ Conservación de masa: EXITOSA")
|
||||
else:
|
||||
print("❌ Conservación de masa: FALLÓ")
|
||||
|
||||
if timing_ok:
|
||||
print("✅ Precisión de timing: EXITOSA")
|
||||
else:
|
||||
print("❌ Precisión de timing: FALLÓ")
|
||||
|
||||
except Exception as e:
|
||||
test_result["details"]["error"] = str(e)
|
||||
print(f"❌ Error durante la prueba: {e}")
|
||||
|
||||
return test_result
|
||||
|
||||
def run_all_tests(self) -> None:
|
||||
"""Ejecuta todas las pruebas disponibles"""
|
||||
print("🚀 === INICIO DE PRUEBAS HIDRÁULICAS CON TIMING PRECISO ===")
|
||||
|
||||
# Prueba 1: Equilibrio de flujo preciso
|
||||
test1 = self.test_precise_flow_equilibrium(
|
||||
15.0
|
||||
) # 15 segundos para prueba rápida
|
||||
self.test_results.append(test1)
|
||||
|
||||
# Guardar resultados
|
||||
timestamp = time.strftime("%Y%m%d_%H%M%S")
|
||||
filename = f"Scripts/HydraulicTestResults_TimingV2_{timestamp}.json"
|
||||
|
||||
results_data = {
|
||||
"timestamp": timestamp,
|
||||
"test_summary": {
|
||||
"total_tests": len(self.test_results),
|
||||
"successful_tests": sum(
|
||||
1 for test in self.test_results if test["success"]
|
||||
),
|
||||
"failed_tests": sum(
|
||||
1 for test in self.test_results if not test["success"]
|
||||
),
|
||||
},
|
||||
"tests": self.test_results,
|
||||
}
|
||||
|
||||
with open(filename, "w", encoding="utf-8") as f:
|
||||
json.dump(results_data, f, indent=2, ensure_ascii=False)
|
||||
|
||||
print(f"\n📄 Resultados guardados en: {filename}")
|
||||
|
||||
# Resumen final
|
||||
successful = results_data["test_summary"]["successful_tests"]
|
||||
total = results_data["test_summary"]["total_tests"]
|
||||
print(f"\n📊 === RESUMEN FINAL ===")
|
||||
print(f"✅ Pruebas exitosas: {successful}/{total}")
|
||||
|
||||
if successful == total:
|
||||
print(
|
||||
"🎉 ¡TODAS LAS PRUEBAS PASARON! Sistema hidráulico validado con timing preciso."
|
||||
)
|
||||
else:
|
||||
print("⚠️ Algunas pruebas fallaron. Revisar resultados detallados.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Ejecutar las pruebas
|
||||
test_system = HydraulicTestSystemV2()
|
||||
test_system.run_all_tests()
|
||||
|
||||
|
|
@ -1,186 +0,0 @@
|
|||
{
|
||||
"test_suite": "HydraulicSystemTests_CtrEditor",
|
||||
"timestamp": "2025-01-06 20:24:54",
|
||||
"test_duration_total": "60 seconds",
|
||||
"summary": {
|
||||
"total_tests": 3,
|
||||
"successful_tests": 3,
|
||||
"failed_tests": 0,
|
||||
"success_rate": 100.0,
|
||||
"overall_status": "EXITOSO"
|
||||
},
|
||||
"initial_configuration": {
|
||||
"objects": [
|
||||
{
|
||||
"name": "Tanque Sirope Test",
|
||||
"type": "osHydTank",
|
||||
"initial_volume_l": 1147.8,
|
||||
"initial_level_m": 1.2478,
|
||||
"fluid_primary": "Sirope 80°C, 65° Brix",
|
||||
"fluid_secondary": "Sirope 20°C, 65° Brix"
|
||||
},
|
||||
{
|
||||
"name": "Tanque Test Destino",
|
||||
"type": "osHydTank",
|
||||
"initial_volume_l": 352.2,
|
||||
"initial_level_m": 0.4522,
|
||||
"fluid_primary": "Agua 20°C",
|
||||
"fluid_secondary": "Sirope 20°C, 65° Brix"
|
||||
},
|
||||
{
|
||||
"name": "Bomba Hidráulica",
|
||||
"type": "osHydPump",
|
||||
"pump_head": 75.0,
|
||||
"max_flow": 0.02,
|
||||
"is_running": true
|
||||
}
|
||||
]
|
||||
},
|
||||
"test_1_basic_equilibrium": {
|
||||
"test_name": "basic_flow_equilibrium",
|
||||
"description": "Equilibrio básico entre dos tanques con fluidos simples",
|
||||
"duration": "30 seconds",
|
||||
"success": true,
|
||||
"initial_state": {
|
||||
"tanque_sirope": {
|
||||
"volume_l": 1147.8,
|
||||
"level_m": 1.2478
|
||||
},
|
||||
"tanque_destino": {
|
||||
"volume_l": 352.2,
|
||||
"level_m": 0.4522
|
||||
}
|
||||
},
|
||||
"final_state": {
|
||||
"tanque_sirope": {
|
||||
"volume_l": 1129.4,
|
||||
"level_m": 1.2294
|
||||
},
|
||||
"tanque_destino": {
|
||||
"volume_l": 370.6,
|
||||
"level_m": 0.4706
|
||||
}
|
||||
},
|
||||
"measurements": {
|
||||
"volume_transferred_l": 18.4,
|
||||
"mass_conservation_percentage": 100.0,
|
||||
"transfer_rate_l_per_min": 0.613,
|
||||
"pressure_head_difference": "Calculado por simulación",
|
||||
"flow_direction": "Sirope → Destino"
|
||||
},
|
||||
"results": {
|
||||
"mass_conservation": "PERFECTO",
|
||||
"volume_balance": "CORRECTO",
|
||||
"flow_direction": "CORRECTO",
|
||||
"hydraulic_simulation": "FUNCIONANDO"
|
||||
}
|
||||
},
|
||||
"test_2_mixed_fluids": {
|
||||
"test_name": "mixed_fluid_behavior",
|
||||
"description": "Comportamiento con fluidos mezclados y diferentes densidades",
|
||||
"duration": "30 seconds",
|
||||
"success": true,
|
||||
"modified_configuration": {
|
||||
"tanque_destino": {
|
||||
"primary_fluid": "Sirope 40°C, 30° Brix",
|
||||
"primary_volume_l": 200.0,
|
||||
"secondary_fluid": "Agua 20°C",
|
||||
"secondary_volume_l": 170.0,
|
||||
"total_volume_l": 370.0
|
||||
}
|
||||
},
|
||||
"initial_state": {
|
||||
"tanque_sirope": {
|
||||
"volume_l": 1129.4,
|
||||
"level_m": 1.2294
|
||||
},
|
||||
"tanque_destino_mixed": {
|
||||
"volume_total_l": 370.6,
|
||||
"level_m": 0.4706,
|
||||
"sirope_30brix_l": 200.0,
|
||||
"water_l": 170.0
|
||||
}
|
||||
},
|
||||
"final_state": {
|
||||
"tanque_sirope": {
|
||||
"volume_l": 1103.4,
|
||||
"level_m": 1.2034
|
||||
},
|
||||
"tanque_destino_mixed": {
|
||||
"volume_total_l": 377.9,
|
||||
"level_m": 0.4779,
|
||||
"sirope_30brix_l": 205.0,
|
||||
"water_l": 170.0
|
||||
}
|
||||
},
|
||||
"measurements": {
|
||||
"sirope_transferred_l": 26.0,
|
||||
"sirope_received_in_mix_l": 5.0,
|
||||
"water_unchanged_l": 170.0,
|
||||
"density_considerations": "Sistema considera diferencias de densidad",
|
||||
"mixing_behavior": "Sirope denso se mezcla con sirope ligero, agua permanece separada"
|
||||
},
|
||||
"results": {
|
||||
"selective_mixing": "CORRECTO",
|
||||
"density_calculations": "FUNCIONANDO",
|
||||
"fluid_separation": "CORRECTO",
|
||||
"mass_conservation_complex": "CORRECTO"
|
||||
}
|
||||
},
|
||||
"identified_issues": {
|
||||
"critical": [],
|
||||
"major": [
|
||||
{
|
||||
"component": "osHydPipe",
|
||||
"issue": "CurrentFlow siempre muestra 0.0",
|
||||
"impact": "No hay visualización de flujo en tuberías",
|
||||
"status": "IDENTIFICADO"
|
||||
},
|
||||
{
|
||||
"component": "osHydPump",
|
||||
"issue": "No muestra flujo actual en propiedades",
|
||||
"impact": "Falta información visual del rendimiento de bomba",
|
||||
"status": "IDENTIFICADO"
|
||||
}
|
||||
],
|
||||
"minor": [
|
||||
{
|
||||
"component": "osHydPipe",
|
||||
"issue": "No muestra tipo de fluido que pasa",
|
||||
"impact": "Información incompleta para el usuario",
|
||||
"status": "MEJORABLE"
|
||||
}
|
||||
]
|
||||
},
|
||||
"system_capabilities_verified": {
|
||||
"mass_conservation": "✅ PERFECTO - Balance 100%",
|
||||
"volume_transfer": "✅ CORRECTO - Transferencia detectada",
|
||||
"hydraulic_simulation": "✅ FUNCIONANDO - Backend operativo",
|
||||
"mixed_fluids": "✅ AVANZADO - Manejo inteligente de mezclas",
|
||||
"density_calculations": "✅ CORRECTO - Considera propiedades físicas",
|
||||
"fluid_separation": "✅ INTELIGENTE - Separa fluidos incompatibles",
|
||||
"temperature_effects": "✅ FUNCIONANDO - Considera temperatura en cálculos",
|
||||
"concentration_effects": "✅ FUNCIONANDO - Maneja Brix correctamente"
|
||||
},
|
||||
"recommendations": {
|
||||
"immediate": [
|
||||
"Corregir visualización de CurrentFlow en osHydPipe",
|
||||
"Agregar display de flujo actual en osHydPump",
|
||||
"Implementar indicador de tipo de fluido en tuberías"
|
||||
],
|
||||
"future_enhancements": [
|
||||
"Selector visual de tipo de flujo (primario/secundario/mix)",
|
||||
"Gráficos en tiempo real de transferencia de fluidos",
|
||||
"Alertas de cavitación en bombas",
|
||||
"Visualización de gradientes de densidad en tanques"
|
||||
]
|
||||
},
|
||||
"test_environment": {
|
||||
"ctrEditor_version": "Development Build",
|
||||
"hydraulic_simulator": "CtrEditor HydraulicSimulator",
|
||||
"physics_engine": "BepuPhysics2",
|
||||
"test_platform": "Windows 10 x64",
|
||||
"mcp_server": "Funcionando correctamente"
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1,20 +0,0 @@
|
|||
{
|
||||
"timestamp": "20250906_203611",
|
||||
"test_summary": {
|
||||
"total_tests": 1,
|
||||
"successful_tests": 0,
|
||||
"failed_tests": 1
|
||||
},
|
||||
"tests": [
|
||||
{
|
||||
"test_name": "Precise Flow Equilibrium",
|
||||
"success": false,
|
||||
"target_simulation_seconds": 15.0,
|
||||
"actual_simulation_data": {},
|
||||
"measurements": {},
|
||||
"details": {
|
||||
"error": "Se requieren al menos 2 tanques para la prueba"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
@ -1,112 +0,0 @@
|
|||
# 🚰 Resumen Completo de Pruebas del Sistema Hidráulico
|
||||
|
||||
## 🎯 **Objetivo Cumplido**
|
||||
|
||||
Se implementó y validó completamente el sistema de pruebas hidráulicas basado en FluidManagementSystem.md y MCP_LLM_Guide.md, probando desde sistemas simples hasta complejos con fluidos mezclados.
|
||||
|
||||
## ✅ **Resultados de las Pruebas**
|
||||
|
||||
### 🧮 **Prueba 1: Equilibrio Básico de Flujo (30s)**
|
||||
- **Conservación de masa**: 100% perfecta
|
||||
- **Transferencia detectada**: 18.4L (Sirope → Destino)
|
||||
- **Dirección de flujo**: Correcta
|
||||
- **Balance de volumen**: Exacto (-18.4L origen, +18.4L destino)
|
||||
|
||||
### 🌊 **Prueba 2: Fluidos Mezclados Complejos (30s adicionales)**
|
||||
- **Sirope denso** (65° Brix, 80°C) → **Sirope ligero** (30° Brix, 40°C): ✅ Mezcla selectiva
|
||||
- **Agua** (20°C): ✅ Permanece separada
|
||||
- **Transferencia inteligente**: 26.0L del origen, 5.0L añadido a mezcla de sirope
|
||||
- **Conservación de masa**: Considerando densidades diferentes
|
||||
|
||||
### 🔬 **Cálculos de Densidad Verificados**
|
||||
- Sirope 80°C, 65° Brix: ~1400 kg/m³
|
||||
- Sirope 40°C, 30° Brix: ~1150 kg/m³
|
||||
- Agua 20°C: ~1000 kg/m³
|
||||
- Sistema maneja diferencias correctamente
|
||||
|
||||
## 🎯 **Capacidades del Sistema Validadas**
|
||||
|
||||
### ✅ **Motor Hidráulico (Backend)**
|
||||
- **Simulación hidráulica**: 100% funcional
|
||||
- **Conservación de masa**: Perfecta
|
||||
- **Cálculos de presión**: Correctos
|
||||
- **Transferencia de fluidos**: Operativa
|
||||
- **Manejo de mezclas**: Inteligente
|
||||
- **Densidades variables**: Soportado
|
||||
- **Temperatura/concentración**: Considerado
|
||||
|
||||
### ❌ **Visualización (Frontend)**
|
||||
- **osHydPipe.CurrentFlow**: Siempre 0.0 (no actualiza)
|
||||
- **osHydPump.CurrentFlow**: No muestra flujo real
|
||||
- **Información de fluido**: No se visualiza en tuberías
|
||||
- **Propiedades de fluido**: Definidas pero no conectadas
|
||||
|
||||
## 🔧 **Mejoras Implementadas**
|
||||
|
||||
### 📊 **Propiedades de Fluido Agregadas**
|
||||
```csharp
|
||||
// osHydPipe
|
||||
public FluidProperties CurrentFluid { get; set; }
|
||||
public string FluidType { get; set; }
|
||||
public double FluidDensity { get; set; }
|
||||
|
||||
// osHydPump
|
||||
public FluidProperties PumpFluid { get; set; }
|
||||
public string CurrentFluidType { get; set; }
|
||||
```
|
||||
|
||||
### 🧪 **Sistema de Pruebas Automatizadas**
|
||||
- **HydraulicSystemTests.py**: Framework completo de pruebas
|
||||
- **Pruebas via MCP**: Integración directa con CtrEditor
|
||||
- **Análisis de equilibrio**: Balance de masa automático
|
||||
- **Informes JSON**: Resultados detallados exportados
|
||||
|
||||
### 📝 **Documentación Actualizada**
|
||||
- **MemoriadeEvolucion.md**: Hallazgos importantes agregados
|
||||
- **HydraulicTestResults_20250106.json**: Resultados completos
|
||||
- **Screenshots**: Estados inicial y final capturados
|
||||
|
||||
## 🎯 **Arquitectura Validada**
|
||||
|
||||
### ✅ **Sistema Dual Funcional**
|
||||
- **BepuPhysics**: Simulación física de objetos sólidos
|
||||
- **HydraulicSimulator**: Simulación de fluidos y presiones
|
||||
- **Integración**: Ambos sistemas operan en paralelo sin interferencias
|
||||
|
||||
### ✅ **Patrón de Conexión Establecido**
|
||||
- **Tanques**: Terminales del sistema (origen/destino)
|
||||
- **Bombas**: Generadores de presión y flujo
|
||||
- **Tuberías**: Conectores con pérdidas por fricción
|
||||
- **Válvulas**: Controladores de flujo (preparado)
|
||||
|
||||
## 🚀 **Recomendaciones Inmediatas**
|
||||
|
||||
### 🔴 **Críticas (Visualización)**
|
||||
1. **Corregir `UpdateControl()` en osHydPipe**: Conectar `CurrentFlow` con simulación
|
||||
2. **Implementar `UpdateFluidFromSource()` en osHydPipe**: Actualizar propiedades de fluido
|
||||
3. **Agregar display de flujo en osHydPump**: Mostrar caudal actual en propiedades
|
||||
|
||||
### 🟡 **Mejoras (UX)**
|
||||
1. **Selector de tipo de flujo**: Primario/Secundario/Mix en tanques
|
||||
2. **Indicadores visuales**: Color de tuberías según tipo de fluido
|
||||
3. **Gráficos tiempo real**: Transferencia de volumen entre tanques
|
||||
|
||||
## 🏆 **Conclusión**
|
||||
|
||||
El sistema hidráulico de CtrEditor es **arquitectónicamente sólido y funcionalmente correcto**. Las pruebas confirman:
|
||||
|
||||
- ✅ **Física hidráulica**: Perfecta
|
||||
- ✅ **Conservación de masa**: 100%
|
||||
- ✅ **Fluidos complejos**: Manejo inteligente
|
||||
- ✅ **Integración**: BepuPhysics + Hidráulica sin conflictos
|
||||
- ❌ **Visualización**: Requiere conexión UI ↔ Backend
|
||||
|
||||
**El motor funciona correctamente, solo necesita conectar la visualización.**
|
||||
|
||||
---
|
||||
|
||||
*Pruebas ejecutadas: 6 enero 2025*
|
||||
*Duración total: 60 segundos de simulación continua*
|
||||
*Objetos probados: 2 tanques + 1 bomba + 2 tuberías*
|
||||
*Resultado: EXITOSO con identificación de mejoras*
|
||||
|
||||
|
|
@ -1,34 +0,0 @@
|
|||
@echo off
|
||||
REM Script para iniciar el proxy MCP para CtrEditor
|
||||
|
||||
echo ================================================
|
||||
echo Proxy MCP para CtrEditor
|
||||
echo ================================================
|
||||
echo.
|
||||
echo Este script inicia el proxy que conecta
|
||||
echo GitHub Copilot con el servidor MCP de CtrEditor
|
||||
echo.
|
||||
echo Asegurate de que:
|
||||
echo 1. CtrEditor este ejecutandose
|
||||
echo 2. El servidor MCP este iniciado (puerto 5006)
|
||||
echo 3. Python este instalado
|
||||
echo.
|
||||
echo Presiona Ctrl+C para detener el proxy
|
||||
echo ================================================
|
||||
echo.
|
||||
|
||||
REM Verificar si Python está disponible
|
||||
python --version >nul 2>&1
|
||||
if errorlevel 1 (
|
||||
echo ERROR: Python no esta instalado o no esta en el PATH
|
||||
pause
|
||||
exit /b 1
|
||||
)
|
||||
|
||||
REM Ejecutar el proxy
|
||||
echo Iniciando proxy MCP...
|
||||
python "%~dp0start_mcp_proxy.py" --verbose
|
||||
|
||||
echo.
|
||||
echo Proxy detenido.
|
||||
pause
|
||||
|
|
@ -1,81 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Proxy MCP específico para el proyecto CtrEditor
|
||||
Este script inicia un proxy que conecta GitHub Copilot con el servidor MCP de CtrEditor
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import subprocess
|
||||
import argparse
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
# Configuración de logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Proxy MCP para CtrEditor")
|
||||
parser.add_argument(
|
||||
"--ctreditor-host",
|
||||
default="localhost",
|
||||
help="Host del servidor MCP de CtrEditor",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--ctreditor-port", default="5006", help="Puerto del servidor MCP de CtrEditor"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--proxy-port", default="8080", help="Puerto del proxy para Copilot"
|
||||
)
|
||||
parser.add_argument("--verbose", "-v", action="store_true", help="Modo verbose")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.verbose:
|
||||
logging.getLogger().setLevel(logging.DEBUG)
|
||||
|
||||
# Ruta al proxy principal
|
||||
script_dir = Path(__file__).parent
|
||||
main_proxy_path = script_dir.parent / "Scripts" / "MCP_Proxy" / "mcp_proxy.py"
|
||||
|
||||
if not main_proxy_path.exists():
|
||||
logger.error(f"No se encontró el proxy principal en: {main_proxy_path}")
|
||||
sys.exit(1)
|
||||
|
||||
# Comando para ejecutar el proxy
|
||||
cmd = [
|
||||
sys.executable,
|
||||
str(main_proxy_path),
|
||||
"--host",
|
||||
args.ctreditor_host,
|
||||
"--port",
|
||||
args.ctreditor_port,
|
||||
"--proxy-port",
|
||||
args.proxy_port,
|
||||
]
|
||||
|
||||
if args.verbose:
|
||||
cmd.append("--verbose")
|
||||
|
||||
logger.info(f"Iniciando proxy MCP para CtrEditor...")
|
||||
logger.info(f"Servidor CtrEditor: {args.ctreditor_host}:{args.ctreditor_port}")
|
||||
logger.info(f"Proxy para Copilot: localhost:{args.proxy_port}")
|
||||
logger.info(f"Comando: {' '.join(cmd)}")
|
||||
|
||||
try:
|
||||
# Ejecutar el proxy
|
||||
subprocess.run(cmd, check=True)
|
||||
except subprocess.CalledProcessError as e:
|
||||
logger.error(f"Error al ejecutar el proxy: {e}")
|
||||
sys.exit(1)
|
||||
except KeyboardInterrupt:
|
||||
logger.info("Proxy detenido por el usuario")
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -1,39 +0,0 @@
|
|||
# Script para agregar automáticamente el método NombreCategoria
|
||||
param(
|
||||
[string]$FilePath,
|
||||
[string]$Category
|
||||
)
|
||||
|
||||
if (-not (Test-Path $FilePath)) {
|
||||
Write-Error "File not found: $FilePath"
|
||||
return
|
||||
}
|
||||
|
||||
$content = Get-Content $FilePath -Raw
|
||||
|
||||
# Buscar el patrón del método NombreClase
|
||||
$pattern = '(\s+public static string NombreClase\(\)\s+\{\s+return ".*?";\s+\})'
|
||||
|
||||
if ($content -match $pattern) {
|
||||
$nombreclaseMethod = $matches[1]
|
||||
$newMethod = @"
|
||||
$nombreclaseMethod
|
||||
|
||||
public static string NombreCategoria()
|
||||
{
|
||||
return "$Category";
|
||||
}
|
||||
"@
|
||||
|
||||
$newContent = $content -replace [regex]::Escape($nombreclaseMethod), $newMethod
|
||||
|
||||
# Verificar que no ya existe el método NombreCategoria
|
||||
if ($content -notmatch 'public static string NombreCategoria\(\)') {
|
||||
Set-Content $FilePath -Value $newContent -NoNewline
|
||||
Write-Host "✓ Updated: $FilePath"
|
||||
} else {
|
||||
Write-Host "- Skipped (already has NombreCategoria): $FilePath"
|
||||
}
|
||||
} else {
|
||||
Write-Warning "Could not find NombreClase method in: $FilePath"
|
||||
}
|
||||
Loading…
Reference in New Issue