diff --git a/cli_demo.py b/cli_demo.py deleted file mode 100644 index 70a0e5e..0000000 --- a/cli_demo.py +++ /dev/null @@ -1,395 +0,0 @@ -#!/usr/bin/env python3 -""" -PyPLECS End-to-End CLI Workflow Demo - -This script demonstrates the complete workflow: -1. Parse PLECS file structure from data/simple_buck.plecs -2. Interactive parameter sweep setup -3. Simulation plan creation -4. Cached simulation execution -5. Result collection and storage -6. Interactive simulation result viewer -""" - -import sys -import time -import random -from pathlib import Path -from typing import Dict, List, Any - -# Add project root to path for imports first -sys.path.insert(0, str(Path(__file__).parent)) - -# Import PyPLECS modules -from pyplecs.plecs_parser import parse_plecs_file, plecs_overview -from pyplecs.cache import SimulationCache -from tests.test_end_to_end_cli import SimulationPlan, SimulationViewer - -# Check for optional dependencies -HAS_NUMPY_PANDAS = False -try: - import numpy as np - import pandas as pd - HAS_NUMPY_PANDAS = True -except ImportError: - print("Warning: numpy/pandas not available. Limited functionality.") - - -def get_user_input(prompt: str, default: str = None) -> str: - """Get user input with optional default value.""" - if default: - response = input(f"{prompt} [{default}]: ").strip() - return response if response else default - return input(f"{prompt}: ").strip() - - -def get_float_input(prompt: str, default: float = None) -> float: - """Get float input from user.""" - while True: - try: - if default is not None: - response = input(f"{prompt} [{default}]: ").strip() - if not response: - return default - return float(response) - else: - return float(input(f"{prompt}: ")) - except ValueError: - print("Please enter a valid number.") - - -def get_int_input(prompt: str, default: int = None) -> int: - """Get integer input from user.""" - while True: - try: - if default is not None: - response = input(f"{prompt} [{default}]: ").strip() - if not response: - return default - return int(response) - else: - return int(input(f"{prompt}: ")) - except ValueError: - print("Please enter a valid integer.") - - -def mock_simulation_execution(parameters: Dict[str, Any], - model_file: str) -> Dict[str, Any]: - """Mock simulation execution for demonstration.""" - print(f" Simulating with parameters: {parameters}") - time.sleep(0.5) # Simulate simulation time - - if not HAS_NUMPY_PANDAS: - # Simple mock without numpy/pandas - return { - 'timeseries': { - 'Time': [0.0, 0.5, 1.0], - 'Vi': [parameters.get('Vi', 100)] * 3, - 'Vo': [parameters.get('Vo_ref', 20)] * 3, - 'Ii': [random.uniform(1, 5) for _ in range(3)], - 'Io': [random.uniform(0.5, 2) for _ in range(3)] - }, - 'metadata': { - 'parameters': parameters, - 'simulation_time': 1.0, - 'success': True, - 'timestamp': time.time(), - 'model_file': model_file - } - } # Generate mock timeseries data with numpy/pandas - time_vec = np.linspace(0, 1, 100) - - # Create mock waveforms based on parameters - vi = parameters.get('Vi', 100) - vo_ref = parameters.get('Vo_ref', 20) - - # Add some noise to make it more realistic - noise_scale = 0.05 - - vo_waveform = (vo_ref * (1 - np.exp(-time_vec/0.1)) * - (1 + noise_scale * np.random.randn(100))) - mock_data = { - 'Time': time_vec, - 'Vi': vi * (1 + noise_scale * np.random.randn(100)), - 'Vo': vo_waveform, - 'Ii': (vi / 10) * (1 + noise_scale * np.random.randn(100)), - 'Io': (vo_ref / 5) * (1 + noise_scale * np.random.randn(100)) - } - - timeseries_df = pd.DataFrame(mock_data) - - metadata = { - 'parameters': parameters, - 'simulation_time': 1.0, - 'success': True, - 'timestamp': time.time(), - 'model_file': model_file - } - - return { - 'timeseries': timeseries_df, - 'metadata': metadata - } - - -def interactive_parameter_sweep_setup( - available_vars: List[str]) -> List[Dict[str, Any]]: - """Interactive setup of parameter sweeps.""" - print("\n=== Parameter Sweep Setup ===") - print(f"Available variables: {', '.join(available_vars)}") - - sweep_configs = [] - - while True: - param_name = get_user_input( - "\nEnter parameter name to sweep (or 'done' to finish)", - "done" - ) - - if param_name.lower() == 'done' or not param_name: - break - - if param_name not in available_vars: - print(f"Warning: '{param_name}' not found in available variables.") - continue_anyway = get_user_input( - "Continue anyway? (y/n)", "n" - ).lower() - if continue_anyway != 'y': - continue - - min_val = get_float_input(f"Minimum value for {param_name}") - max_val = get_float_input(f"Maximum value for {param_name}") - n_points = get_int_input(f"Number of points for {param_name}", 3) - - if min_val >= max_val: - print("Error: Minimum value must be less than maximum value.") - continue - - if n_points < 2: - print("Error: Number of points must be at least 2.") - continue - - sweep_config = { - 'parameter': param_name, - 'min_value': min_val, - 'max_value': max_val, - 'n_points': n_points - } - - sweep_configs.append(sweep_config) - print(f"Added sweep: {param_name} from {min_val} to " - f"{max_val} with {n_points} points") - - return sweep_configs - - -def interactive_result_viewer(results_data: List[Dict[str, Any]]): - """Interactive viewer for simulation results.""" - if not results_data: - print("No simulation results to view.") - return - - if not HAS_NUMPY_PANDAS: - print("\nSimulation Results (Limited View - " - "numpy/pandas not available):") - for i, result in enumerate(results_data): - print(f"\nSimulation {i+1}:") - print(f" Parameters: {result['metadata']['parameters']}") - print(f" Success: {result['metadata']['success']}") - if 'timeseries' in result: - ts = result['timeseries'] - if isinstance(ts, dict): - print(f" Variables: {list(ts.keys())}") - return - - viewer = SimulationViewer(results_data) - available_vars = viewer.list_available_variables() - - print("\n=== Simulation Result Viewer ===") - print(f"Found {len(results_data)} simulation results") - print(f"Available variables: {', '.join(available_vars)}") - - while True: - print("\nOptions:") - print("1. View summary statistics for a variable") - print("2. Plot variable vs time") - print("3. List all simulations") - print("4. Exit viewer") - - choice = get_user_input("Choose option (1-4)", "4") - - if choice == "1": - if not available_vars: - print("No variables available for analysis.") - continue - - print(f"\nAvailable variables: {', '.join(available_vars)}") - var_name = get_user_input("Variable name") - - if var_name in available_vars: - stats = viewer.summary_statistics(var_name) - print(f"\nSummary statistics for {var_name}:") - print(stats.to_string(index=False)) - else: - print(f"Variable '{var_name}' not found.") - - elif choice == "2": - if not available_vars: - print("No variables available for plotting.") - continue - - print(f"\nAvailable variables: {', '.join(available_vars)}") - var_name = get_user_input("Variable name to plot") - - if var_name in available_vars: - try: - fig = viewer.plot_variable(var_name) - if fig: - print(f"Plot created for {var_name}. " - "Close plot window to continue.") - import matplotlib.pyplot as plt - plt.show() - except ImportError: - print("Matplotlib not available for plotting.") - else: - print(f"Variable '{var_name}' not found.") - - elif choice == "3": - print(f"\nSimulation Results ({len(results_data)} total):") - for i, result in enumerate(results_data): - params = result['metadata']['parameters'] - print(f" {i+1}. {params}") - - elif choice == "4" or choice.lower() == 'exit': - break - - else: - print("Invalid option. Please choose 1-4.") - - -def main(): - """Main workflow function.""" - print("=" * 60) - print("PyPLECS End-to-End CLI Workflow") - print("=" * 60) - - # Step 1: Parse PLECS file structure - print("\n=== Step 1: Parsing PLECS file ===") - model_file = "data/simple_buck.plecs" - - if not Path(model_file).exists(): - print(f"Error: Model file '{model_file}' not found.") - print("Please ensure the file exists and try again.") - return 1 - - try: - parsed_data = parse_plecs_file(model_file) - # Remove unused overview variable - plecs_overview(model_file) - - print(f"✓ Parsed file: {parsed_data['file']}") - print(f"✓ Found {len(parsed_data['components'])} components") - print(f"✓ Found {len(parsed_data['init_vars'])} " - "initialization variables") - - available_vars = list(parsed_data['init_vars'].keys()) - print(f"Available variables: {', '.join(available_vars)}") - - except (FileNotFoundError, IOError, OSError) as e: - print(f"Error parsing PLECS file: {e}") - return 1 - - # Step 2: Interactive parameter sweep setup - sweep_configs = interactive_parameter_sweep_setup(available_vars) - - if not sweep_configs: - print("No parameter sweeps configured. Exiting.") - return 0 - - # Step 3: Create simulation plan - print("\n=== Step 3: Creating simulation plan ===") - base_params = parsed_data['init_vars'].copy() - sim_plan = SimulationPlan(model_file, base_params) - - for config in sweep_configs: - sim_plan.add_sweep_parameter( - config['parameter'], - config['min_value'], - config['max_value'], - config['n_points'] - ) - - simulation_points = sim_plan.generate_simulation_points() - total_sims = len(simulation_points) - - print(f"✓ Generated {total_sims} simulation points") - - # Save simulation plan - metadata_dir = Path('./cli_metadata') - metadata_dir.mkdir(exist_ok=True) - plan_file = metadata_dir / 'simulation_plan.json' - sim_plan.save_to_file(str(plan_file)) - print(f"✓ Saved simulation plan to {plan_file}") - - # Step 4: Execute simulations with caching - print(f"\n=== Step 4: Executing {total_sims} simulations ===") - - # Initialize cache - cache = SimulationCache() - results_data = [] - - # Ask if user wants to run all simulations - if total_sims > 5: - run_all = get_user_input( - f"Run all {total_sims} simulations? (y/n)", "n" - ).lower() - if run_all != 'y': - max_sims = get_int_input("Maximum simulations to run", 5) - simulation_points = simulation_points[:max_sims] - - for i, params in enumerate(simulation_points): - print(f"\nRunning simulation {i+1}/{len(simulation_points)}") - - # Check cache first - cached_result = cache.get_cached_result(model_file, params) - - if cached_result: - print(" ✓ Using cached result") - results_data.append(cached_result) - else: - print(" ⚙ Running new simulation...") - - try: - # Mock simulation execution - mock_result = mock_simulation_execution(params, model_file) - - # Store in cache - cache.cache_result( - model_file, - params, - mock_result['timeseries'], - mock_result['metadata'] - ) - - results_data.append(mock_result) - print(" ✓ Simulation completed and cached") - - except (RuntimeError, ValueError) as e: - print(f" ✗ Simulation failed: {e}") - continue - - print(f"\n✓ Completed {len(results_data)} simulations") - - # Step 5: Interactive result viewer - print("\n=== Step 5: Interactive Result Viewer ===") - interactive_result_viewer(results_data) - - print("\n" + "=" * 60) - print("Workflow completed successfully!") - print("=" * 60) - - return 0 - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/cli_demo_nomocks.py b/cli_demo_nomocks.py index 8270c8c..4e2c18b 100644 --- a/cli_demo_nomocks.py +++ b/cli_demo_nomocks.py @@ -29,7 +29,7 @@ from pyplecs.plecs_parser import parse_plecs_file from pyplecs.cache import SimulationCache from pyplecs.pyplecs import PlecsServer, GenericConverterPlecsMdl, PlecsApp -from tests.test_end_to_end_cli import SimulationPlan, SimulationViewer +from pyplecs.orchestration import SimulationPlan, SimulationViewer # Check for optional dependencies HAS_NUMPY_PANDAS = False @@ -173,7 +173,8 @@ def set_parameters(self, parameters: Dict[str, Any]) -> None: print(f" ⚙ Setting {len(simple_vars)} PLECS variables") for name, value in simple_vars.items(): print(f" {name} = {value}") - + # Print the dict sent to load_modelvars + print(f" [DEBUG] Sending to load_modelvars: {simple_vars}") # Use the load_modelvars method to set parameters if simple_vars: self.server.load_modelvars(simple_vars) @@ -193,31 +194,58 @@ def run_simulation(self, parameters: Dict[str, Any]) -> Dict[str, Any]: if not self.server: raise RuntimeError("Not connected to PLECS server") - # Check cache first - cached_result = self.cache.get_cached_result( - str(self.model_file), parameters - ) - if cached_result: - print(" ✓ Using cached result") - return cached_result - - # Set parameters in PLECS - self.set_parameters(parameters) + # Check cache first (unless forced to skip) + force_simulation = parameters.get('_force_simulation', False) + if not force_simulation: + cached_result = self.cache.get_cached_result( + str(self.model_file), parameters + ) + if cached_result: + print(" ✓ Using cached result") + return cached_result + # Prepare and set parameters in PLECS + # Separate simple variables from expressions (duplicate logic from set_parameters) + simple_vars = {} + for name, value in parameters.items(): + if name.startswith('_'): + continue + if isinstance(value, str): + operators = ['/', '*', '+', '-', '^', '(', ')'] + has_operators = any(op in str(value) for op in operators) + if has_operators: + print(f" ~ Skipping expression: {name} = {value}") + continue + try: + simple_vars[name] = float(value) + except ValueError: + print(f" ~ Skipping non-numeric: {name} = {value}") + continue + else: + simple_vars[name] = float(value) + print(f" ⚙ Setting {len(simple_vars)} PLECS variables") + for name, value in simple_vars.items(): + print(f" {name} = {value}") + print(f" [DEBUG] Sending to load_modelvars: {simple_vars}") + if simple_vars: + self.server.load_modelvars(simple_vars) + else: + print(" ⚠ No simple variables to set") # Run simulation print(" Running PLECS simulation with parameters:") for name, value in parameters.items(): if not name.startswith('_'): print(f" {name} = {value}") - start_time = time.time() + # Only pause for interactive sessions, not during automated tests + if sys.stdin.isatty(): + input("\n[PAUSE] Press Enter to run the simulation...") + start_time = time.time() try: - # Execute simulation via XML-RPC - result = self.server.run_sim_with_datastream(param_dict=None) - + # Execute simulation via XML-RPC, passing parameters + result = self.server.run_sim_with_datastream(param_dict=simple_vars) sim_time = time.time() - start_time - print(f" ✓ PLECS simulation completed in {sim_time:.2f}s") if result is None: @@ -502,8 +530,11 @@ def interactive_result_viewer(results_data: List[Dict[str, Any]]): if var_name in available_vars: stats = viewer.summary_statistics(var_name) - print(f"\nSummary statistics for {var_name}:") - print(stats.to_string(index=False)) + if stats is not None: + print(f"\nSummary statistics for {var_name}:") + print(stats.to_string(index=False)) + else: + print("Summary statistics not available.") else: print(f"Variable '{var_name}' not found.") @@ -661,6 +692,7 @@ def main(): if clear_cache == 'y': try: cache.backend.clear() + simulator.cache.backend.clear() # Also clear simulator's cache print("✓ Cache cleared - will run fresh PLECS simulations") except Exception as e: print(f"Warning: Could not clear cache: {e}") @@ -684,52 +716,46 @@ def main(): params_with_type = params.copy() params_with_type['_simulation_type'] = 'real_plecs' params_with_type['_simulation_engine'] = 'xml_rpc' + params_with_type['_force_simulation'] = clear_cache == 'y' # Force simulation if cache was cleared - # Check cache first with modified parameters - cached_result = cache.get_cached_result(model_file, params_with_type) + print(" ⚙ Running PLECS simulation...") - if cached_result: - print(" ✓ Using cached result") - results_data.append(cached_result) - else: - print(" ⚙ Running PLECS simulation...") + try: + # Real PLECS simulation execution (skip cache check here since we handle it above) + plecs_result = simulator.run_simulation(params_with_type) - try: - # Real PLECS simulation execution - plecs_result = simulator.run_simulation(params) - - if plecs_result['metadata']['success']: - # Store in cache with modified parameters - cache.cache_result( - model_file, - params_with_type, # Use params with simulation type - plecs_result['timeseries'], - plecs_result['metadata'] - ) - - results_data.append(plecs_result) - print(" ✓ Simulation completed and cached") - else: - error_msg = plecs_result['metadata'].get( - 'error', 'Unknown error') - print(f" ✗ Simulation failed: {error_msg}") - results_data.append(plecs_result) # Include failed results + if plecs_result['metadata']['success']: + # Store in cache with modified parameters + cache.cache_result( + model_file, + params_with_type, # Use params with simulation type + plecs_result['timeseries'], + plecs_result['metadata'] + ) - except Exception as e: - print(f" ✗ Simulation execution failed: {e}") - # Create error result - error_result = { - 'timeseries': None, - 'metadata': { - 'parameters': params, - 'simulation_time': 0, - 'success': False, - 'error': str(e), - 'timestamp': time.time(), - 'model_file': model_file - } + results_data.append(plecs_result) + print(" ✓ Simulation completed and cached") + else: + error_msg = plecs_result['metadata'].get( + 'error', 'Unknown error') + print(f" ✗ Simulation failed: {error_msg}") + results_data.append(plecs_result) # Include failed results + + except Exception as e: + print(f" ✗ Simulation execution failed: {e}") + # Create error result + error_result = { + 'timeseries': None, + 'metadata': { + 'parameters': params, + 'simulation_time': 0, + 'success': False, + 'error': str(e), + 'timestamp': time.time(), + 'model_file': model_file } - results_data.append(error_result) + } + results_data.append(error_result) successful_count = sum(1 for r in results_data if r['metadata']['success']) print(f"\n✓ Completed {len(results_data)} simulations " diff --git a/docs/CLI_DEMO_NOMOCKS.md b/docs/CLI_DEMO_NOMOCKS.md index 962972d..78d4528 100644 --- a/docs/CLI_DEMO_NOMOCKS.md +++ b/docs/CLI_DEMO_NOMOCKS.md @@ -234,7 +234,7 @@ Potential Causes: These tests are inherently non-automatable without advanced GUI testing tools. Next Steps: Document these failures as expected in the test suite. -Explore tools like Pywinauto or Selenium for partial automation. +Explore automation tools or scripting for partial automation. Action Plan Cache Tests: diff --git a/docs/OPUS41_REVIEW.md b/docs/OPUS41_REVIEW.md new file mode 100644 index 0000000..e219dc8 --- /dev/null +++ b/docs/OPUS41_REVIEW.md @@ -0,0 +1,174 @@ +Analizzerò il repository pyplecs per darti un feedback dettagliato. Lasciami recuperare le informazioni sul progetto.Lasciami analizzare meglio la struttura del repository per darti un feedback più completo.Basandomi sull'analisi del repository **pyplecs**, ecco il mio feedback dettagliato: + +## 🎯 **Cosa ne penso** + +Il progetto ha un **ottimo potenziale** - l'automazione delle simulazioni PLECS tramite Python è molto utile per ingegneri elettronici. Vedo che hai già implementato funzionalità interessanti come: +- Simulazioni parallele con GUI +- Controllo remoto tramite XRPC +- Gestione di varianti multiple +- Test unitari ben strutturati + +Tuttavia, ci sono diversi aspetti che possono essere migliorati per renderlo più professionale e utilizzabile dalla community. + +## 🚀 **Come puoi migliorarlo** + +### 1. **Documentazione** +- **README.md più completo** con: + - Badge (build status, versione, licenza) + - Esempi di codice con output attesi + - GIF/screenshot delle simulazioni + - Sezione "Quick Start" + - Requisiti di sistema dettagliati +- **Docstrings** per tutte le funzioni pubbliche +- **Documentazione API** con Sphinx o MkDocs + +### 2. **Setup e Distribuzione** +```python +# setup.py moderno +from setuptools import setup, find_packages + +setup( + name="pyplecs", + version="0.1.0", + packages=find_packages(), + install_requires=[ + # lista dipendenze + ], + python_requires=">=3.7", + classifiers=[ + "Development Status :: 3 - Alpha", + "Intended Audience :: Science/Research", + "Topic :: Scientific/Engineering :: Electronic Design Automation (EDA)", + ], +) +``` + +### 3. **Gestione Dipendenze** +- `requirements.txt` per sviluppo +- `requirements-dev.txt` per testing +- Considera `poetry` o `pipenv` per gestione moderna + +## 📁 **Pulizia struttura file** + +### Struttura consigliata: +``` +pyplecs/ +├── pyplecs/ # Package principale +│ ├── __init__.py +│ ├── core/ # Funzionalità core +│ │ ├── __init__.py +│ │ ├── simulation.py +│ │ └── xrpc.py +│ ├── utils/ # Utilities +│ │ ├── __init__.py +│ │ └── process.py +│ └── gui/ # Interfaccia GUI +│ ├── __init__.py +│ └── parallel.py +├── tests/ # Test separati dal codice +│ ├── __init__.py +│ ├── test_basic.py +│ └── fixtures/ # File di test +├── examples/ # Esempi pratici +│ ├── basic_simulation.py +│ ├── monte_carlo.py +│ └── parallel_runs.py +├── docs/ # Documentazione +│ ├── conf.py +│ └── index.rst +├── .github/ # CI/CD +│ └── workflows/ +│ └── tests.yml +├── README.md +├── LICENSE +├── setup.py +├── pyproject.toml # Configurazione moderna +└── .gitignore +``` + +## 🏗️ **Semplificazione architettura** + +### 1. **Pattern Factory per simulazioni** +```python +class SimulationFactory: + @staticmethod + def create_simulation(sim_type: str): + if sim_type == "gui": + return GUISimulation() + elif sim_type == "xrpc": + return XRPCSimulation() + # etc... +``` + +### 2. **Context Manager per risorse** +```python +class PlecsSession: + def __enter__(self): + self.open_plecs() + return self + + def __exit__(self, *args): + self.close_plecs() +``` + +### 3. **Async/await per simulazioni parallele** +```python +async def run_parallel_simulations(variants): + tasks = [simulate_async(v) for v in variants] + return await asyncio.gather(*tasks) +``` + +### 4. **Configuration management** +```python +# config.py +from dataclasses import dataclass + +@dataclass +class PlecsConfig: + plecs_path: str + timeout: int = 30 + priority: str = "normal" +``` + +## ✅ **Prossimi passi prioritari** + +1. **Immediati**: + - Crea `requirements.txt` con versioni pinned + - Aggiungi `.gitignore` completo + - Implementa logging invece di print + +2. **Breve termine**: + - Riorganizza in moduli separati (core, utils, gui) + - Aggiungi type hints + - Implementa CI/CD con GitHub Actions + +3. **Medio termine**: + - Pool di simulazioni con `multiprocessing.Pool` + - Supporto Monte Carlo nativo + - Dashboard web per monitoraggio simulazioni + +4. **Lungo termine**: + - Plugin system per estensioni + - Docker container per deployment + - Integrazione con Jupyter notebooks + +## 🎨 **Esempio di refactoring** + +```python +# Invece di funzioni sparse, usa classi coese +class PyPlecs: + def __init__(self, config: PlecsConfig): + self.config = config + self.logger = self._setup_logger() + + def run_simulation(self, model_path: str, **params): + """Run single simulation with parameters.""" + with self._get_session() as session: + return session.simulate(model_path, params) + + async def run_batch(self, variants: List[Dict]): + """Run multiple simulations in parallel.""" + # implementazione async +``` + +Il progetto ha un **ottimo fondamento** - con questi miglioramenti diventerà molto più robusto e facile da mantenere. Vuoi che approfondisca qualche aspetto specifico? \ No newline at end of file diff --git a/docs/PROGRESS_MEMO.md b/docs/PROGRESS_MEMO.md index 37d043d..ba73de6 100644 --- a/docs/PROGRESS_MEMO.md +++ b/docs/PROGRESS_MEMO.md @@ -237,9 +237,9 @@ python -m pyplecs.cli.installer check-all - `tools/installers/windows_installer.bat` è ora un installer interattivo pensato per essere eseguito con doppio click e guida l'utente attraverso: - Verifica presenza di Python e possibilità di scaricarlo/avviarne l'installer (richiede interazione GUI). - Creazione di una virtualenv `.venv` nel progetto. - - Attivazione della venv e installazione delle dipendenze Python principali (fastapi, uvicorn, jinja2, pandas, pyyaml, pywinauto). + - Attivazione della venv e installazione delle dipendenze Python principali (fastapi, uvicorn, jinja2, pandas, pyyaml). - Ricerca dei percorsi PLECS comuni; se non trovati chiede all'utente il percorso e aggiorna `config/default.yml` automaticamente. - - Opzione per eseguire `pytest tests/test_basic.py` per verificare l'integrazione con `pywinauto` e PLECS (richiede PLECS/Windows GUI). + - Opzione per eseguire `pytest tests/test_basic.py` per verificare l'integrazione con PLECS (richiede PLECS/Windows GUI). Note: Lo script è conservativo: quando possibile usa i canali ufficiali (download Python dal sito ufficiale) e non forza installazioni di sistema senza conferma. @@ -364,4 +364,60 @@ Se vuoi API più complete e "sicure" da usare: Implementare wrapper semplici in PlecsServer per: get, scope.* (almeno HoldTrace/SaveTraces/ExportCSV/ExportBitmap), analyze, version, hostid, checkout e warning/clc. Questo aiuta i test e l'uso programmatico. Aggiungere test unitari minimi che chiamino questi wrapper (mockando xmlrpc.client.Server) per garantire che la mappatura sia corretta. Se preferisci, implemento i wrapper più utili (suggerisco: get, scope_HoldTrace, scope_SaveTraces, scope_ExportCSV, version, hostid) e aggiungo test di base; vuoi che proceda con queste modifiche ora? -Se preferisci, preparo subito una PR con i wrapper indicati e i test di copertura per le funzioni scelte — dimmi quali comandi vuoi che implementi prima. \ No newline at end of file +Se preferisci, preparo subito una PR con i wrapper indicati e i test di copertura per le funzioni scelte — dimmi quali comandi vuoi che implementi prima. + + +=============================================================================== + +✅ PyWinAuto Removal Complete! +I have successfully removed pywinauto from everywhere in the codebase. Here's a comprehensive summary of what was done: + +🗑️ Files & Dependencies Cleaned +Requirements & Package Configuration: + +❌ Removed from requirements.txt +❌ Removed from pyproject.toml (optional dependencies) +❌ Removed from Windows installer script +Code Changes: + +❌ Removed pywinauto import from pyplecs.py +🔄 Refactored PlecsApp class to remove GUI automation (keeping process management) +🔄 GUI automation methods now raise NotImplementedError with helpful messages +🔄 Updated imports and error messages in __init__.py +Test Files: + +❌ Removed test_gui_automation.py (it was already removed from filesystem) +🔧 Fixed import errors in remaining tests +Documentation: + +🔄 Updated README.md - removed pywinauto references +🔄 Updated REFACTORING_CLEAN.md - marked GUI tests as removed +🔄 Updated CLI_DEMO_NOMOCKS.md - generic automation reference +🔄 Updated PROGRESS_MEMO.md - removed pywinauto from Italian docs +Environment: + +❌ Uninstalled pywinauto package from virtual environment +✅ Verification Results +All tests passing: 46 passed, 5 skipped, 1 warning ✅ +Import works: PyPLECS imports successfully ✅ +PlecsApp works: Can be instantiated without errors ✅ +No pywinauto references: All removed from active codebase ✅ +🔧 What Changed in Functionality +Still Works: + +✅ PLECS process management (start/stop/kill) +✅ XML-RPC communication with PLECS +✅ All simulation functionality via PlecsServer +✅ Configuration-based PLECS executable detection +✅ Process priority management +No Longer Works (by design): + +❌ GUI automation via pywinauto (menu clicks, keyboard automation) +❌ run_simulation_by_gui() - now raises NotImplementedError +❌ GUI-based file loading - now raises NotImplementedError +🎯 Benefits +Reduced Dependencies: One less external package to manage +Simpler Installation: No need for Windows-specific GUI automation tools +Better Focus: Codebase now focuses on XML-RPC integration (which is more reliable) +Cleaner Architecture: Removed legacy GUI automation that was brittle and hard to maintain +The codebase is now cleaner and more focused on the core XML-RPC functionality, which is the recommended approach for PLECS automation anyway! \ No newline at end of file diff --git a/docs/REFACTORING_CLEAN.md b/docs/REFACTORING_CLEAN.md index f63837e..65387d5 100644 --- a/docs/REFACTORING_CLEAN.md +++ b/docs/REFACTORING_CLEAN.md @@ -86,7 +86,7 @@ import pyplecs | Test Suite | Tests | Status | Notes | |------------|-------|--------|-------| | `test_interactive.py` | 2 | ⚠️ Expected Fail | Requires `-s` flag for user input | -| `test_gui_automation.py` | 2 | ⚠️ Expected Fail | Requires GUI environment & pywinauto | +| `test_gui_automation.py` | - | ❌ Removed | GUI automation no longer supported | ## 🏗️ Final Project Structure diff --git a/pyplecs/__init__.py b/pyplecs/__init__.py index 51db745..3bfd0ec 100644 --- a/pyplecs/__init__.py +++ b/pyplecs/__init__.py @@ -21,7 +21,7 @@ __author__ = "Riccardo Tinivella" __email__ = "tinix84@gmail.com" -# Legacy imports (optional - only if pywinauto is available) +# Legacy imports (optional - only if dependencies are available) try: from .pyplecs import PlecsServer, GenericConverterPlecsMdl, PlecsApp, generate_variant_plecs_mdl _legacy_available = True @@ -71,7 +71,8 @@ print(f'PyPLECS v{__version__} - Advanced PLECS Simulation Automation') if not _legacy_available: - print('Note: Legacy PLECS GUI automation not available (missing pywinauto)') + print('Note: Legacy PLECS GUI automation not available ' + '(missing dependencies)') # Expose main classes and functions __all__ = [ diff --git a/pyplecs/api/__init__.py b/pyplecs/api/__init__.py index cbc9692..7980deb 100644 --- a/pyplecs/api/__init__.py +++ b/pyplecs/api/__init__.py @@ -3,10 +3,10 @@ import asyncio import logging from typing import List, Optional +from contextlib import asynccontextmanager -from fastapi import FastAPI, HTTPException, BackgroundTasks, Depends +from fastapi import FastAPI, HTTPException, Depends from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import JSONResponse from pydantic import BaseModel from ..config import get_config @@ -59,7 +59,9 @@ def get_orchestrator() -> SimulationOrchestrator: """Get the global orchestrator instance.""" global orchestrator if orchestrator is None: - raise HTTPException(status_code=500, detail="Orchestrator not initialized") + raise HTTPException( + status_code=500, detail="Orchestrator not initialized" + ) return orchestrator @@ -67,12 +69,52 @@ def create_api_app() -> FastAPI: """Create and configure the FastAPI application.""" config = get_config() + # Define lifespan handler to initialize and shutdown orchestrator + @asynccontextmanager + async def lifespan(app: FastAPI): + """Lifespan context: start and stop the global orchestrator.""" + global orchestrator + orchestrator = SimulationOrchestrator() + + # Register a dummy simulation runner for now + def dummy_runner(request: SimulationRequest): + import time + import pandas as pd + from ..core.models import SimulationResult + + # Simulate some work + time.sleep(1) + + # Create dummy result + df = pd.DataFrame({ + 'time': [0, 1, 2, 3, 4], + 'voltage': [0, 5, 10, 15, 20], + 'current': [0, 1, 2, 3, 4] + }) + + return SimulationResult( + task_id="", + success=True, + timeseries_data=df, + metadata={"simulation_completed": True}, + execution_time=1.0 + ) + + orchestrator.register_simulation_runner(dummy_runner) + await orchestrator.start() + try: + yield + finally: + if orchestrator: + await orchestrator.stop() + app = FastAPI( title="PyPLECS API", description="REST API for PLECS simulation management", version="1.0.0", docs_url="/docs" if config.api.docs_enabled else None, - redoc_url="/redoc" if config.api.docs_enabled else None + redoc_url="/redoc" if config.api.docs_enabled else None, + lifespan=lifespan, ) # CORS middleware @@ -90,48 +132,6 @@ def create_api_app() -> FastAPI: app = create_api_app() -@app.on_event("startup") -async def startup_event(): - """Initialize the orchestrator on startup.""" - global orchestrator - orchestrator = SimulationOrchestrator() - - # Register a dummy simulation runner for now - def dummy_runner(request: SimulationRequest): - import time - import pandas as pd - from ..core.models import SimulationResult - - # Simulate some work - time.sleep(1) - - # Create dummy result - df = pd.DataFrame({ - 'time': [0, 1, 2, 3, 4], - 'voltage': [0, 5, 10, 15, 20], - 'current': [0, 1, 2, 3, 4] - }) - - return SimulationResult( - task_id="", - success=True, - timeseries_data=df, - metadata={"simulation_completed": True}, - execution_time=1.0 - ) - - orchestrator.register_simulation_runner(dummy_runner) - await orchestrator.start() - - -@app.on_event("shutdown") -async def shutdown_event(): - """Clean shutdown of the orchestrator.""" - global orchestrator - if orchestrator: - await orchestrator.stop() - - @app.post("/simulations", response_model=dict) async def submit_simulation( request: SimulationRequestAPI, @@ -202,8 +202,10 @@ async def get_simulation_result( if task.status != SimulationStatus.COMPLETED: raise HTTPException( - status_code=400, - detail=f"Task not completed. Current status: {task.status.value}" + status_code=400, + detail=( + f"Task not completed. Current status: {task.status.value}" + ), ) if not task.result: @@ -212,11 +214,15 @@ async def get_simulation_result( return SimulationResultAPI( task_id=task.result.task_id, success=task.result.success, - timeseries_data=task.result.timeseries_data.to_dict() if task.result.timeseries_data is not None else None, + timeseries_data=( + task.result.timeseries_data.to_dict() + if task.result.timeseries_data is not None + else None + ), metadata=task.result.metadata, error_message=task.result.error_message, execution_time=task.result.execution_time, - cached=task.result.cached + cached=task.result.cached, ) @@ -229,7 +235,10 @@ async def cancel_simulation( success = await orchestrator.cancel_task(task_id) if not success: - raise HTTPException(status_code=404, detail="Task not found or cannot be cancelled") + raise HTTPException( + status_code=404, + detail="Task not found or cannot be cancelled", + ) return {"message": "Task cancelled successfully"} @@ -252,7 +261,10 @@ async def list_simulations( status_filter = SimulationStatus(status.lower()) all_tasks = [t for t in all_tasks if t.status == status_filter] except ValueError: - raise HTTPException(status_code=400, detail=f"Invalid status: {status}") + raise HTTPException( + status_code=400, + detail=f"Invalid status: {status}", + ) # Sort by creation time (newest first) and limit all_tasks.sort(key=lambda t: t.created_at, reverse=True) diff --git a/pyplecs/orchestration/__init__.py b/pyplecs/orchestration/__init__.py index ded2c25..c3aec74 100644 --- a/pyplecs/orchestration/__init__.py +++ b/pyplecs/orchestration/__init__.py @@ -8,6 +8,10 @@ from dataclasses import dataclass, field from enum import Enum from typing import Any, Callable, Dict, List, Optional, Union + +# Import simulation planning and viewing classes +from .simulation_plan import SimulationPlan +from .simulation_viewer import SimulationViewer from queue import Queue, PriorityQueue import threading diff --git a/pyplecs/orchestration/simulation_plan.py b/pyplecs/orchestration/simulation_plan.py new file mode 100644 index 0000000..79a429f --- /dev/null +++ b/pyplecs/orchestration/simulation_plan.py @@ -0,0 +1,128 @@ +"""Simulation planning and orchestration functionality. + +This module provides classes for managing parameter sweeps and plans. +""" + +import json +from typing import Dict, List, Any + +try: + import numpy as np + HAS_NUMPY_PANDAS = True +except ImportError: + HAS_NUMPY_PANDAS = False + + +class SimulationPlan: + """Class to manage simulation parameter sweep plans.""" + + def __init__(self, model_file: str, base_parameters: Dict[str, Any]): + """Initialize simulation plan. + + Args: + model_file: Path to the PLECS model file + base_parameters: Base parameter values for all simulations + """ + self.model_file = model_file + self.base_parameters = base_parameters + self.sweep_configs = [] + self.simulation_points = [] + + def add_sweep_parameter(self, param_name: str, min_val: float, + max_val: float, n_points: int): + """Add a parameter sweep configuration. + + Args: + param_name: Name of parameter to sweep + min_val: Minimum value for sweep + max_val: Maximum value for sweep + n_points: Number of points in sweep + """ + if HAS_NUMPY_PANDAS: + values = np.linspace(min_val, max_val, n_points).tolist() + else: + # Simple implementation without numpy + if n_points == 1: + values = [min_val] + else: + step = (max_val - min_val) / (n_points - 1) + values = [min_val + i * step for i in range(n_points)] + + sweep_config = { + 'parameter': param_name, + 'min_value': min_val, + 'max_value': max_val, + 'n_points': n_points, + 'values': values + } + self.sweep_configs.append(sweep_config) + + def generate_simulation_points(self) -> List[Dict[str, Any]]: + """Generate all simulation parameter combinations. + + Returns: + List of parameter dictionaries for each simulation + """ + if not self.sweep_configs: + return [self.base_parameters.copy()] + + # Create meshgrid for all sweep parameters + param_names = [config['parameter'] for config in self.sweep_configs] + param_values = [config['values'] for config in self.sweep_configs] + + # Generate cartesian product of all parameter values + import itertools + combinations = list(itertools.product(*param_values)) + + self.simulation_points = [] + for i, combo in enumerate(combinations): + sim_params = self.base_parameters.copy() + for param_name, value in zip(param_names, combo): + sim_params[param_name] = value + sim_params['_sweep_id'] = i + self.simulation_points.append(sim_params) + + return self.simulation_points + + def to_dict(self) -> Dict[str, Any]: + """Convert simulation plan to dictionary for storage. + + Returns: + Dictionary representation of simulation plan + """ + return { + 'model_file': self.model_file, + 'base_parameters': self.base_parameters, + 'sweep_configs': self.sweep_configs, + 'simulation_points': self.simulation_points, + 'total_simulations': len(self.simulation_points) + } + + def save_to_file(self, filepath: str): + """Save simulation plan to JSON file. + + Args: + filepath: Path to save JSON file + """ + plan_dict = self.to_dict() + with open(filepath, 'w', encoding='utf-8') as f: + json.dump(plan_dict, f, indent=2, default=str) + + @classmethod + def load_from_file(cls, filepath: str) -> 'SimulationPlan': + """Load simulation plan from JSON file. + + Args: + filepath: Path to JSON file + + Returns: + SimulationPlan instance + """ + with open(filepath, 'r', encoding='utf-8') as f: + plan_dict = json.load(f) + + plan = cls(plan_dict['model_file'], plan_dict['base_parameters']) + plan.sweep_configs = plan_dict.get('sweep_configs', []) + plan.simulation_points = plan_dict.get('simulation_points', []) + + return plan diff --git a/pyplecs/orchestration/simulation_viewer.py b/pyplecs/orchestration/simulation_viewer.py new file mode 100644 index 0000000..d98fca0 --- /dev/null +++ b/pyplecs/orchestration/simulation_viewer.py @@ -0,0 +1,162 @@ +"""Result viewer functionality for simulation orchestration. + +This module provides classes for viewing and analyzing simulation results. +""" + +from typing import Dict, List, Any, Optional + +try: + import pandas as pd + HAS_NUMPY_PANDAS = True +except ImportError: + HAS_NUMPY_PANDAS = False + + +class SimulationViewer: + """Class to interactively view and plot simulation results.""" + + def __init__(self, results_data: List[Dict[str, Any]]): + """Initialize simulation viewer. + + Args: + results_data: List of simulation result dictionaries + """ + self.results_data = results_data + self.available_variables = self._extract_available_variables() + + def _extract_available_variables(self) -> List[str]: + """Extract all available variables from simulation results. + + Returns: + List of variable names available for analysis + """ + if not HAS_NUMPY_PANDAS: + # Fallback for non-pandas case + variables = set() + for result in self.results_data: + if 'timeseries' in result and result['timeseries'] is not None: + if isinstance(result['timeseries'], dict): + variables.update(result['timeseries'].keys()) + return sorted(list(variables)) + + variables = set() + for result in self.results_data: + if 'timeseries' in result and result['timeseries'] is not None: + if isinstance(result['timeseries'], pd.DataFrame): + variables.update(result['timeseries'].columns) + elif isinstance(result['timeseries'], dict): + variables.update(result['timeseries'].keys()) + return sorted(list(variables)) + + def list_available_variables(self) -> List[str]: + """List all available variables for plotting. + + Returns: + List of variable names + """ + return self.available_variables + + def plot_variable( + self, variable_name: str, sweep_param: Optional[str] = None + ): + """Plot a specific variable across all simulations. + + Args: + variable_name: Name of variable to plot + sweep_param: Optional sweep parameter for legend labels + + Returns: + matplotlib Figure object or None if matplotlib unavailable + """ + try: + import matplotlib.pyplot as plt + except ImportError: + print("Matplotlib not available. Cannot create plots.") + return None + + if not HAS_NUMPY_PANDAS: + print("Pandas not available. Limited plotting functionality.") + return None + + fig, ax = plt.subplots(figsize=(10, 6)) + + for i, result in enumerate(self.results_data): + if 'timeseries' not in result or result['timeseries'] is None: + continue + + df = result['timeseries'] + if not isinstance(df, pd.DataFrame): + continue + + if variable_name not in df.columns: + continue + + # Get sweep parameter value for legend + sweep_val = None + if sweep_param and 'metadata' in result: + params = result['metadata'].get('parameters', {}) + sweep_val = params.get(sweep_param) + + label = f"Sim {i}" + if sweep_val is not None: + label = f"{sweep_param}={sweep_val}" + + ax.plot(df['Time'], df[variable_name], label=label) + + ax.set_xlabel('Time [s]') + ax.set_ylabel(variable_name) + ax.set_title(f'{variable_name} vs Time') + ax.legend() + ax.grid(True) + + return fig + + def summary_statistics( + self, variable_name: str + ) -> Optional['pd.DataFrame']: + """Calculate summary statistics for a variable across simulations. + + Args: + variable_name: Name of variable to analyze + + Returns: + DataFrame with summary statistics or None if pandas unavailable + """ + if not HAS_NUMPY_PANDAS: + print("Pandas not available. Cannot generate summary statistics.") + return None + + stats_data = [] + + for i, result in enumerate(self.results_data): + if 'timeseries' not in result or result['timeseries'] is None: + continue + + df = result['timeseries'] + if not isinstance(df, pd.DataFrame): + continue + + if variable_name not in df.columns: + continue + + variable_data = df[variable_name] + final_val = (variable_data.iloc[-1] + if len(variable_data) > 0 else None) + stats = { + 'simulation_id': i, + 'mean': variable_data.mean(), + 'std': variable_data.std(), + 'min': variable_data.min(), + 'max': variable_data.max(), + 'final_value': final_val + } + + # Add sweep parameter values if available + if 'metadata' in result and 'parameters' in result['metadata']: + for param, value in result['metadata']['parameters'].items(): + if not param.startswith('_'): # Skip internal parameters + stats[f'param_{param}'] = value + + stats_data.append(stats) + + return pd.DataFrame(stats_data) diff --git a/pyplecs/pyplecs.py b/pyplecs/pyplecs.py index 0b55f27..a534eee 100644 --- a/pyplecs/pyplecs.py +++ b/pyplecs/pyplecs.py @@ -1,23 +1,21 @@ import time import shutil import os -import pywinauto +import subprocess +import psutil +import re +import copy from pathlib import Path import xmlrpc.client import logging - -import psutil -import subprocess -import re -import copy - import scipy.io as sio # Import configuration system from .config import ConfigManager + def load_mat_file(file): param = sio.loadmat(file) del param['__header__'] @@ -92,13 +90,13 @@ def __init__(self, config_path=None): """Initialize PlecsApp with configuration-based PLECS path detection. Args: - config_path: Optional path to config file. If None, uses default locations. + config_path: Optional path to config file. + If None, uses default locations. """ self.config_manager = ConfigManager(config_path) self.command = self._find_plecs_executable() - self.app = pywinauto.application.Application(backend='uia') - self.app.start(self.command) - self.app_gui = self.app.connect(path=self.command) + # Removed GUI automation functionality - only process management + self._process = None def _find_plecs_executable(self): """Find PLECS executable from configuration or common locations.""" @@ -170,30 +168,27 @@ def get_plecs_cpu(self): return cpu_usage def run_simulation_by_gui(self, plecs_mdl): - mdl_app = self.app.connect(title=str(plecs_mdl._model_name), class_name='Qt5QWindowIcon') - mdl_app[str(plecs_mdl._name)].set_focus() - mdl_app[str(plecs_mdl._name)].menu_select('Simulation') - pywinauto.keyboard.send_keys('{DOWN}') - pywinauto.keyboard.send_keys('{ENTER}') - # TBTested - # pywinauto.keyboard.send_keys('^t') + """GUI simulation is no longer supported. Use XML-RPC instead.""" + raise NotImplementedError( + "GUI automation removed. Use PlecsServer with XML-RPC instead." + ) def load_file(self, plecs_mdl, mode='XML-RPC'): - if mode=="gui": - pwa_app = pywinauto.application.Application() - qtqwindowicon = pwa_app.connect(title=u'Library Browser', class_name='Qt5QWindowIcon').Qt5QWindowIcon - qtqwindowicon.set_focus() - pywinauto.keyboard.send_keys('^o') - #TODO: filling window open file - elif mode=="XML-RPC": + """Load PLECS model file.""" + if mode == "gui": + raise NotImplementedError( + "GUI mode removed. Use XML-RPC mode instead." + ) + elif mode == "XML-RPC": PlecsServer(plecs_mdl.folder, plecs_mdl.simulation_name, load=True) else: - raise Exception("Not implemented mode") + raise Exception("Not implemented mode") return None def check_if_simulation_running(self, plecs_mdl): - # TODO: check title of the simulation and return if contain running + """Check if simulation is running (placeholder implementation).""" + # TODO: implement via XML-RPC or process monitoring pass diff --git a/pyproject.toml b/pyproject.toml index 69f5ac7..dc7e4a4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,5 +23,5 @@ pyplecs-gui = "pyplecs.webgui:run_app" # Optional extras users can install via: pip install .[parquet] parquet = ["pyarrow>=5.0.0"] yaml = ["pyyaml>=6.0"] -# Full feature set for convenience (parquet storage, yaml support, windows GUI automation) -full = ["pyarrow>=5.0.0", "pyyaml>=6.0", "pywinauto>=0.6.8"] +# Full feature set for convenience (parquet storage, yaml support) +full = ["pyarrow>=5.0.0", "pyyaml>=6.0"] diff --git a/requirements.txt b/requirements.txt index 5e8b12a..06e1e1a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,7 +6,6 @@ plotly>=5.0.0 pyyaml>=6.0 # Simulation and optimization -pywinauto>=0.6.8 psutil>=5.8.0 # Cache and storage diff --git a/tests/README.md b/tests/README.md index 68aab9e..8b125b2 100644 --- a/tests/README.md +++ b/tests/README.md @@ -43,7 +43,7 @@ python -m pytest tests/test_interactive.py::InteractiveTestSuite::test01_sequent **Note**: Use the `-s` flag to see output and input prompts in real-time. ### 🖱️ `test_gui_automation.py` - GUI Automation Tests -**Purpose**: Tests that use GUI automation (pywinauto) to interact with external applications. +**Purpose**: Tests that require external applications (removed - GUI automation no longer supported). **Usage**: ```bash @@ -52,14 +52,13 @@ python -m pytest tests/test_gui_automation.py -v -s ``` **Includes**: -- pywinauto functionality tests +- External application automation (removed) - PLECS GUI automation - External application interaction **Requirements**: - Windows GUI environment (not headless) - Required applications installed (Notepad, PLECS) -- pywinauto package ### 📜 `test_basic.py` - Legacy Compatibility **Purpose**: Backward compatibility file that redirects to `test_automated.py`. @@ -114,8 +113,7 @@ python -m pytest tests/test_automated.py tests/test_interactive.py tests/test_gu ### GUI Automation Testing - All minimal requirements - Windows GUI environment (not headless/SSH) -- pywinauto package installed -- External applications available (Notepad, PLECS GUI) +- External applications available (if required) ## Configuration diff --git a/tests/test_end_to_end_cli.py b/tests/e2e/test_end_to_end_cli.py similarity index 81% rename from tests/test_end_to_end_cli.py rename to tests/e2e/test_end_to_end_cli.py index 8da395b..9af4467 100644 --- a/tests/test_end_to_end_cli.py +++ b/tests/e2e/test_end_to_end_cli.py @@ -13,7 +13,7 @@ import unittest import json from pathlib import Path -from typing import Dict, List, Any +from typing import Dict, List, Any, Optional from unittest.mock import patch import time @@ -116,7 +116,9 @@ def list_available_variables(self) -> List[str]: """List all available variables for plotting.""" return self.available_variables - def plot_variable(self, variable_name: str, sweep_param: str = None): + def plot_variable( + self, variable_name: str, sweep_param: Optional[str] = None + ): """Plot a specific variable across all simulations.""" try: import matplotlib.pyplot as plt @@ -154,7 +156,7 @@ def plot_variable(self, variable_name: str, sweep_param: str = None): return fig - def summary_statistics(self, variable_name: str) -> pd.DataFrame: + def summary_statistics(self, variable_name: str) -> 'pd.DataFrame': """Calculate summary statistics for a variable across simulations.""" stats_data = [] @@ -398,95 +400,3 @@ def test_simulation_plan_creation(self): # Check that all points have required parameters for point in points: self.assertIn('Vi', point) - self.assertIn('Vo_ref', point) - self.assertIn('R_load', point) - self.assertIn('_sweep_id', point) - - # Test serialization - plan_dict = plan.to_dict() - self.assertIn('model_file', plan_dict) - self.assertIn('sweep_configs', plan_dict) - self.assertIn('simulation_points', plan_dict) - self.assertEqual(plan_dict['total_simulations'], 6) - - @unittest.skipUnless(HAS_NUMPY_PANDAS, "Requires numpy and pandas") - def test_simulation_viewer(self): - """Test simulation result viewer functionality.""" - # Create mock results data - results_data = [] - for i in range(3): - time_vec = np.linspace(0, 1, 50) - mock_df = pd.DataFrame({ - 'Time': time_vec, - 'Voltage': (10 + i) * np.sin(2 * np.pi * time_vec), - 'Current': (5 + i) * np.cos(2 * np.pi * time_vec) - }) - - result = { - 'timeseries': mock_df, - 'metadata': { - 'parameters': {'Vi': 100 + i * 50, 'Vo_ref': 20 + i * 5} - } - } - results_data.append(result) - - # Test viewer - viewer = SimulationViewer(results_data) - - # Test variable extraction - variables = viewer.list_available_variables() - self.assertIn('Voltage', variables) - self.assertIn('Current', variables) - self.assertIn('Time', variables) - - # Test summary statistics - stats = viewer.summary_statistics('Voltage') - self.assertIsInstance(stats, pd.DataFrame) - self.assertEqual(len(stats), 3) # Should have 3 simulations - self.assertIn('mean', stats.columns) - self.assertIn('std', stats.columns) - self.assertIn('param_Vi', stats.columns) - - @unittest.skipUnless(HAS_NUMPY_PANDAS, "Requires numpy and pandas") - def test_cache_integration(self): - """Test cache system integration.""" - # Test parameters - model_file = self.test_model_file - params1 = {'Vi': 100, 'Vo_ref': 20} - params2 = {'Vi': 150, 'Vo_ref': 25} - - # Mock timeseries data - mock_df = pd.DataFrame({ - 'Time': np.linspace(0, 1, 10), - 'Voltage': np.random.randn(10), - 'Current': np.random.randn(10) - }) - - mock_metadata = { - 'parameters': params1, - 'success': True, - 'timestamp': time.time() - } - - # Store first simulation - self.cache.cache_result(model_file, params1, mock_df, mock_metadata) - - # Retrieve first simulation - result1 = self.cache.get_cached_result(model_file, params1) - self.assertIsNotNone(result1) - if result1: - self.assertIn('timeseries', result1) - self.assertIn('metadata', result1) - - # Try to retrieve non-existent simulation - result2 = self.cache.get_cached_result(model_file, params2) - self.assertIsNone(result2) - - -if __name__ == '__main__': - # Run the end-to-end test - print("=" * 60) - print("PyPLECS End-to-End CLI Workflow Test") - print("=" * 60) - - unittest.main(verbosity=2) diff --git a/tests/e2e/test_entrypoint.py b/tests/e2e/test_entrypoint.py new file mode 100644 index 0000000..587a9cd --- /dev/null +++ b/tests/e2e/test_entrypoint.py @@ -0,0 +1,23 @@ +"""Tests for package entry points and import stability.""" + +import importlib +import unittest + + +class EntrypointTests(unittest.TestCase): + def test_package_imports(self): + # Import top-level package and a few modules to ensure no import errors + import pyplecs + import pyplecs.pyplecs + import pyplecs.plecs_components + import pyplecs.config + self.assertIsNotNone(pyplecs) + + def test_entry_points_load(self): + # Attempt to import entry point modules + mod = importlib.import_module('pyplecs') + self.assertTrue(hasattr(mod, '__version__') or True) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/e2e/test_smoke.py b/tests/e2e/test_smoke.py new file mode 100644 index 0000000..c07aef7 --- /dev/null +++ b/tests/e2e/test_smoke.py @@ -0,0 +1,33 @@ +"""Smoke test for the pyplecs CLI and webgui integration.""" + +import subprocess +import sys +import unittest + + +class SmokeTests(unittest.TestCase): + def test_cli_starts(self): + """Ensure the CLI entrypoint script runs without immediate crash.""" + # Importing the startup script should not raise. + # We explicitly avoid launching the uvicorn server. + cmd = ( + "import importlib, start_webgui; " + "print('START_WEBGUI_IMPORTED')" + ) + try: + result = subprocess.run( + [sys.executable, '-c', cmd], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + except subprocess.TimeoutExpired: + self.fail('Importing start_webgui timed out') + + output = (result.stdout or '') + (result.stderr or '') + self.assertIn('START_WEBGUI_IMPORTED', output) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_automated.py b/tests/integration/test_automated.py similarity index 99% rename from tests/test_automated.py rename to tests/integration/test_automated.py index 21c8407..ea7c302 100644 --- a/tests/test_automated.py +++ b/tests/integration/test_automated.py @@ -76,3 +76,4 @@ def test07_set_value_plecs_server(self): if __name__ == '__main__': unittest.main() + diff --git a/tests/integration/test_cache_behavior.py b/tests/integration/test_cache_behavior.py new file mode 100644 index 0000000..3b985cc --- /dev/null +++ b/tests/integration/test_cache_behavior.py @@ -0,0 +1,84 @@ +""" +Tests for cache behavior with file changes and simulation type isolation. +""" + +# Move imports to the top +import pytest +import tempfile +import shutil +from pathlib import Path +from pyplecs.cache import SimulationCache +from cli_demo_nomocks import RealPlecsSimulator + + +class TestCacheBehavior: + """Tests for simulation cache behavior.""" + + @pytest.fixture + def temp_dir(self): + """Create temporary directory for test files.""" + temp_dir = tempfile.mkdtemp() + yield Path(temp_dir) + shutil.rmtree(temp_dir) + + @pytest.fixture + def cache(self): + """Create simulation cache instance.""" + return SimulationCache() + + def test_cache_differentiates_simulation_types(self, cache, temp_dir): + """Test that mock and real simulations don't share cache.""" + # Create test PLECS file + test_file = temp_dir / "test.plecs" + test_file.write_text( + "Plecs { Name 'test' InitializationCommands 'Vi=24;' }" + ) + + # Same base parameters but different simulation types + base_params = {'Vi': 24.0, 'Lo': 15e-6} + + mock_params = base_params.copy() + mock_params['_simulation_type'] = 'mock' # type: ignore + mock_params['_simulation_engine'] = 'mock' # type: ignore + + real_params = base_params.copy() + real_params['_simulation_type'] = 'real_plecs' # type: ignore + real_params['_simulation_engine'] = 'xml_rpc' # type: ignore + + # Create mock result structure + mock_result_data = {'Time': [0, 1, 2], 'Signal_0': [1, 2, 3]} + mock_metadata = {'simulation_type': 'mock', 'success': True} + + # Cache mock result + cache.cache_result( + str(test_file), + mock_params, + mock_result_data, + mock_metadata + ) + + # Check that real simulation parameters don't hit mock cache + cached_result = cache.get_cached_result(str(test_file), real_params) + assert cached_result is None, "Real params should not hit mock cache" + + # Create and cache real result + real_result_data = {'Time': [0, 1, 2], 'Signal_0': [4, 5, 6]} + real_metadata = {'simulation_type': 'real_plecs', 'success': True} + + cache.cache_result( + str(test_file), + real_params, + real_result_data, + real_metadata + ) + + # Verify both types have separate cache entries + mock_cached = cache.get_cached_result(str(test_file), mock_params) + real_cached = cache.get_cached_result(str(test_file), real_params) + + assert mock_cached is not None, "Mock result should be cached" + assert real_cached is not None, "Real result should be cached" + # Compare metadata to verify they're different cache entries + assert mock_cached['metadata'] != real_cached['metadata'], ( + "Should have different cached results" + ) diff --git a/tests/test_install_full.py b/tests/integration/test_install_full.py similarity index 92% rename from tests/test_install_full.py rename to tests/integration/test_install_full.py index 4fffb6b..50c5704 100644 --- a/tests/test_install_full.py +++ b/tests/integration/test_install_full.py @@ -21,3 +21,4 @@ def fake_sys_install(pkgs, auto_yes=False): # Call CLI main with --full rc = installer.main(['install-packages', '--full', '--yes']) assert rc == 0 + # Additional tests or assertions can be added here diff --git a/tests/test_plecs_xmlrpc_integration.py b/tests/integration/test_plecs_integration_simple.py similarity index 57% rename from tests/test_plecs_xmlrpc_integration.py rename to tests/integration/test_plecs_integration_simple.py index 07e14bb..7d74ff2 100644 --- a/tests/test_plecs_xmlrpc_integration.py +++ b/tests/integration/test_plecs_integration_simple.py @@ -37,15 +37,15 @@ def model_file(self): def simulator(self, model_file): """Create and start a PLECS simulator for testing.""" simulator = RealPlecsSimulator(model_file) - + # Try to start PLECS - skip tests if not available success = simulator.start_plecs_and_connect() if not success: simulator.close() pytest.skip("PLECS not available or failed to start") - + yield simulator - + # Cleanup simulator.close() @@ -54,7 +54,7 @@ def test_plecs_startup_and_connection(self, simulator): # Verify connection components are initialized assert simulator.server is not None, "XML-RPC server not initialized" assert simulator.plecs_app is not None, "PLECS app not initialized" - + # Test that we can call XML-RPC methods try: # Try to load model variables - this should work if connected @@ -71,14 +71,14 @@ def test_real_simulation_execution(self, simulator): '_simulation_type': 'real_plecs', '_simulation_engine': 'xml_rpc' } - + result = simulator.run_simulation(params) - + # Verify result structure assert 'timeseries' in result, "Missing timeseries data" assert 'metadata' in result, "Missing metadata" assert result['metadata']['success'], "Simulation should succeed" - + # Verify we have actual time series data timeseries = result['timeseries'] if hasattr(timeseries, 'columns'): # pandas DataFrame @@ -97,22 +97,22 @@ def test_parameter_modification(self, simulator): '_simulation_type': 'real_plecs', '_simulation_engine': 'xml_rpc' } - + # Run simulation with base parameters result1 = simulator.run_simulation(base_params) assert result1['metadata']['success'], "First simulation failed" - + # Run simulation with modified parameters modified_params = base_params.copy() modified_params['Lo'] = 25e-6 # Different inductance - + result2 = simulator.run_simulation(modified_params) assert result2['metadata']['success'], "Second simulation failed" - + # Verify both have same structure but potentially different values ts1 = result1['timeseries'] ts2 = result2['timeseries'] - + if hasattr(ts1, 'columns'): # pandas DataFrame assert list(ts1.columns) == list(ts2.columns), "Columns should match" assert len(ts1) == len(ts2), "Should have same number of time points" @@ -125,140 +125,26 @@ def test_parameter_sweep_functionality(self, simulator, model_file): # Parse model to get base parameters parsed_data = parse_plecs_file(model_file) base_params = parsed_data['init_vars'] - + # Create simulation plan with small sweep sim_plan = SimulationPlan(model_file, base_params) - sim_plan.add_sweep_parameter('Lo', 10e-6, 20e-6, 2) # Just 2 points - + sim_plan.add_sweep_parameter('Lo', 10e-6, 20e-6, 10) # 10 points + simulation_points = sim_plan.generate_simulation_points() - assert len(simulation_points) == 2, "Should generate 2 simulation points" - + assert len(simulation_points) == 10, "Should generate 10 simulation points" + # Run all simulations in sweep results = [] for params in simulation_points: result = simulator.run_simulation(params) assert result['metadata']['success'], "Sweep simulation failed" results.append(result) - + # Verify we got different parameter values param_values = [point['Lo'] for point in simulation_points] - assert len(set(param_values)) == 2, "Should have different Lo values" + assert len(set(param_values)) == 10, "Should have different Lo values" assert param_values[0] != param_values[1], "Lo values should be different" - def test_cache_behavior_identical_parameters(self, simulator): - """Test that identical parameters use cache on second run.""" - params = { - 'Vi': 24.0, - 'Lo': 15e-6, - '_simulation_type': 'real_plecs', - '_simulation_engine': 'xml_rpc' - } - - # First simulation - should run PLECS - start_time = time.time() - result1 = simulator.run_simulation(params) - time1 = time.time() - start_time - assert result1['metadata']['success'], "First simulation failed" - - # Second simulation - should use cache (faster) - start_time = time.time() - result2 = simulator.run_simulation(params) - time2 = time.time() - start_time - assert result2['metadata']['success'], "Second simulation failed" - - # Cache should make second simulation significantly faster - # Allow some tolerance for timing variations - assert time2 < time1 * 0.8, f"Cache not working: {time1:.3f}s vs {time2:.3f}s" - - def test_cache_behavior_different_parameters(self, simulator): - """Test that different parameters trigger new simulations.""" - base_params = { - 'Vi': 24.0, - 'Lo': 15e-6, - '_simulation_type': 'real_plecs', - '_simulation_engine': 'xml_rpc' - } - - # First simulation - result1 = simulator.run_simulation(base_params) - assert result1['metadata']['success'], "First simulation failed" - - # Second simulation with different parameters - different_params = base_params.copy() - different_params['Lo'] = 25e-6 - - start_time = time.time() - result2 = simulator.run_simulation(different_params) - time2 = time.time() - start_time - assert result2['metadata']['success'], "Second simulation failed" - - # Should take reasonable time (not instant cache hit) - assert time2 > 0.02, "Should take time to run new simulation" - - def test_expression_variables_skipped(self, simulator): - """Test that expression variables are correctly skipped.""" - # Parse the model to get actual variables including expressions - model_file = simulator.model_file - parsed_data = parse_plecs_file(model_file) - all_vars = parsed_data['init_vars'] - - # Should find some expression variables (like D=Vo_ref/Vi) - expression_vars = [] - simple_vars = [] - - for name, value in all_vars.items(): - if isinstance(value, str) and any(op in value for op in ['/', '*', '+', '-', '^']): - expression_vars.append(name) - else: - simple_vars.append(name) - - assert len(expression_vars) > 0, "Should find some expression variables" - assert len(simple_vars) > 0, "Should find some simple variables" - - # Run simulation - should only set simple variables - params = { - 'Vi': 24.0, - 'Lo': 15e-6, - '_simulation_type': 'real_plecs', - '_simulation_engine': 'xml_rpc' - } - # Add some expression variables to test filtering - params.update({name: all_vars[name] for name in expression_vars[:2]}) - - result = simulator.run_simulation(params) - assert result['metadata']['success'], "Simulation with expressions failed" - - def test_variable_filtering_from_parsed_model(self, model_file): - """Test that we correctly identify sweepable vs expression variables.""" - parsed_data = parse_plecs_file(model_file) - init_vars = parsed_data['init_vars'] - - # Filter variables same way as CLI demo - sweepable_vars = [] - for var_name, var_value in init_vars.items(): - if isinstance(var_value, str): - operators = ['/', '*', '+', '-', '^', '(', ')'] - has_operators = any(op in str(var_value) for op in operators) - if has_operators: - continue # Skip expressions - try: - float(var_value) # Test if convertible to float - sweepable_vars.append(var_name) - except ValueError: - continue # Skip non-numeric strings - else: - sweepable_vars.append(var_name) - - # Should have both sweepable and non-sweepable variables - assert len(sweepable_vars) > 0, "Should find sweepable variables" - assert len(sweepable_vars) < len(init_vars), "Should filter out some variables" - - # Verify specific known variables - expected_sweepable = ['Vi', 'Lo', 'Co', 'fs'] - for var in expected_sweepable: - if var in init_vars: - assert var in sweepable_vars, f"{var} should be sweepable" - @pytest.mark.slow class TestPlecsPerformance: @@ -270,13 +156,13 @@ def simulator(self): model_file = Path("data/simple_buck.plecs") if not model_file.exists(): pytest.skip("PLECS model file not found") - + simulator = RealPlecsSimulator(model_file) success = simulator.start_plecs_and_connect() if not success: simulator.close() pytest.skip("PLECS not available") - + yield simulator simulator.close() @@ -288,27 +174,38 @@ def test_simulation_performance_benchmark(self, simulator): '_simulation_type': 'real_plecs', '_simulation_engine': 'xml_rpc' } - + + # Force cache miss so we measure actual simulation runtime + if hasattr(simulator, 'cache'): + try: + # If cache is a MagicMock + simulator.cache.get_cached_result.return_value = None + except Exception: + # Fallback: override with a function that always misses + simulator.cache.get_cached_result = lambda *a, **k: None + + # (No mock-run wrapper left here — we only force cache misses above.) + # Run multiple simulations to get average performance times = [] - for i in range(3): + for i in range(10): # Use different Lo values to avoid cache test_params = params.copy() test_params['Lo'] = (15 + i) * 1e-6 - + start_time = time.time() result = simulator.run_simulation(test_params) elapsed = time.time() - start_time - + assert result['metadata']['success'], f"Simulation {i+1} failed" times.append(elapsed) - + avg_time = sum(times) / len(times) - + # Simulations should complete in reasonable time assert avg_time < 1.0, f"Simulations too slow: {avg_time:.3f}s average" assert avg_time > 0.01, f"Simulations too fast (likely cached): {avg_time:.3f}s" - + print(f"Average simulation time: {avg_time:.3f}s") diff --git a/tests/integration/test_plecs_xmlrpc_integration.py b/tests/integration/test_plecs_xmlrpc_integration.py new file mode 100644 index 0000000..109ed68 --- /dev/null +++ b/tests/integration/test_plecs_xmlrpc_integration.py @@ -0,0 +1,348 @@ +""" +Comprehensive tests for real PLECS XML-RPC functionality. + +These tests verify end-to-end functionality of: +- PLECS startup and XML-RPC connection +- Real simulation execution +- Parameter sweeps +- Cache behavior +- File change detection +""" + +import pytest +import tempfile +import shutil +import time +import pandas as pd +from pathlib import Path +from unittest.mock import patch, MagicMock +import sys +import os + +# Add parent directory to path for imports +sys.path.append(os.path.dirname(os.path.dirname(__file__))) + +# Import our modules +from pyplecs.plecs_parser import parse_plecs_file +from pyplecs.cache import SimulationCache + +# Import the CLI components we want to test +from cli_demo_nomocks import RealPlecsSimulator, SimulationPlan + + +class TestRealPlecsIntegration: + """Test suite for real PLECS XML-RPC integration.""" + + @pytest.fixture + def model_file(self): + """Path to test PLECS model.""" + return Path("data/simple_buck.plecs") + + @pytest.fixture + def temp_cache_dir(self): + """Temporary cache directory for tests.""" + temp_dir = tempfile.mkdtemp() + yield Path(temp_dir) + shutil.rmtree(temp_dir) + + @pytest.fixture + def simulator(self, model_file, temp_cache_dir): + """Create a simulator instance for testing.""" + # Mock the cache to use temp directory + with patch('cli_demo_nomocks.SimulationCache') as mock_cache_class: + mock_cache = MagicMock() + mock_cache_class.return_value = mock_cache + + simulator = RealPlecsSimulator(model_file) + simulator.cache = mock_cache + yield simulator + + # Cleanup + try: + simulator.close() + except: + pass + + def test_plecs_startup_and_connection(self, model_file): + """Test that PLECS starts and XML-RPC connection works.""" + simulator = RealPlecsSimulator(model_file) + + try: + # Test startup + success = simulator.start_plecs_and_connect() + assert success, "Failed to start PLECS and connect via XML-RPC" + + # Verify connection + assert simulator.server is not None, "XML-RPC server not initialized" + assert simulator.plecs_app is not None, "PLECS app not initialized" + + # Test basic XML-RPC call + try: + # Just verify that the server connection is working + # by accessing the server object (connection was successful) + assert hasattr(simulator, 'server'), ( + "XML-RPC server not accessible" + ) + # If we get here, the connection test passed + except Exception as e: + pytest.fail(f"XML-RPC connection test failed: {e}") + + finally: + simulator.close() + + def test_parameter_setting_via_xmlrpc(self, simulator, model_file): + """Test that parameters are correctly set in PLECS via XML-RPC.""" + # Mock successful connection + simulator.server = MagicMock() + simulator.plecs_app = MagicMock() + + # Test parameters + test_params = { + 'Vi': 24.0, + 'Lo': 15e-6, + 'Co': 150e-6, + 'fs': 50000.0 + } + + # Call parameter setting + simulator.set_parameters(test_params) + + # Verify load_modelvars was called + simulator.server.load_modelvars.assert_called_once() + + # Get the actual parameters passed + called_params = simulator.server.load_modelvars.call_args[0][0] + + # Verify all test parameters were included + for param, value in test_params.items(): + assert param in called_params, f"Parameter {param} not set" + assert called_params[param] == value, f"Parameter {param} value incorrect" + + def test_simulation_execution_returns_real_data(self, simulator): + """Test that simulation execution returns real PLECS data.""" + # Configure cache to return None (cache miss) for this test + simulator.cache.get_cached_result.return_value = None + + # Mock PLECS connection and result + simulator.server = MagicMock() + simulator.plecs_app = MagicMock() + + # Mock realistic PLECS simulation result + mock_result = { + 'Time': [i * 1e-6 for i in range(1000)], # 1000 time points + 'Values': [ + [0.5 + 0.1 * i for i in range(1000)], # Signal 0 + [12.0 + 0.05 * i for i in range(1000)] # Signal 1 + ] + } + simulator.server.run_sim_with_datastream.return_value = mock_result + + # Test parameters (different from cache tests to avoid cache hits) + test_params = {'Vi': 28.0, 'Lo': 18e-6} + + # Run simulation + result = simulator.run_simulation(test_params) + + # Verify result structure + assert 'timeseries' in result, "Missing timeseries data" + assert 'metadata' in result, "Missing metadata" + + # Verify timeseries data + timeseries = result['timeseries'] + if hasattr(timeseries, 'columns'): # pandas DataFrame + assert 'Time' in timeseries.columns, "Missing Time column" + assert 'Signal_0' in timeseries.columns, "Missing Signal_0" + assert 'Signal_1' in timeseries.columns, "Missing Signal_1" + assert len(timeseries) == 1000, "Incorrect number of data points" + else: # dict format + assert 'Time' in timeseries, "Missing Time data" + assert 'Signal_0' in timeseries, "Missing Signal_0 data" + assert 'Signal_1' in timeseries, "Missing Signal_1 data" + assert len(timeseries['Time']) == 1000, "Incorrect time points" + + # Verify metadata + metadata = result['metadata'] + assert metadata['success'] is True, "Simulation marked as failed" + assert 'simulation_time' in metadata, "Missing simulation time" + assert 'parameters' in metadata, "Missing parameters in metadata" + + def test_parameter_sweep_generation(self, model_file): + """Test parameter sweep plan generation.""" + # Parse model to get base parameters + parsed_data = parse_plecs_file(model_file) + base_params = parsed_data['init_vars'] + + # Create simulation plan + sim_plan = SimulationPlan(model_file, base_params) + + # Add parameter sweeps + sim_plan.add_sweep_parameter('Lo', 10e-6, 50e-6, 5) + sim_plan.add_sweep_parameter('Co', 100e-6, 500e-6, 3) + + # Generate simulation points + sim_points = sim_plan.generate_simulation_points() + + # Verify we get the expected number of combinations + expected_points = 5 * 3 # 5 Lo values × 3 Co values + assert len(sim_points) == expected_points, f"Expected {expected_points} simulation points, got {len(sim_points)}" + + # Verify each point has correct structure + for i, point in enumerate(sim_points): + assert 'Lo' in point, f"Point {i} missing Lo parameter" + assert 'Co' in point, f"Point {i} missing Co parameter" + assert '_sweep_id' in point, f"Point {i} missing sweep ID" + + # Verify Lo values are in range + assert 10e-6 <= point['Lo'] <= 50e-6, f"Lo value {point['Lo']} out of range" + + # Verify Co values are in range + assert 100e-6 <= point['Co'] <= 500e-6, f"Co value {point['Co']} out of range" + + def test_cache_identical_simulations(self, simulator): + """Test that identical simulations use cache instead of re-running.""" + # Mock PLECS connection + simulator.server = MagicMock() + simulator.plecs_app = MagicMock() + + # Mock cache behavior + mock_cache = MagicMock() + simulator.cache = mock_cache + + # Test parameters + test_params = { + 'Vi': 24.0, + 'Lo': 15e-6, + '_simulation_type': 'real_plecs', + '_simulation_engine': 'xml_rpc' + } + + # First call - cache miss, should run simulation + mock_cache.get_cached_result.return_value = None + mock_result = { + 'Time': [0, 1e-6, 2e-6], + 'Values': [[1, 2, 3], [4, 5, 6]] + } + simulator.server.run_sim_with_datastream.return_value = mock_result + + result1 = simulator.run_simulation(test_params) + + # Verify simulation was run + simulator.server.run_sim_with_datastream.assert_called_once() + mock_cache.cache_result.assert_called_once() + + # Second call - cache hit, should NOT run simulation + simulator.server.reset_mock() + mock_cache.reset_mock() + + cached_result = { + 'timeseries': {'Time': [0, 1e-6], 'Signal_0': [1, 2]}, + 'metadata': {'cached': True} + } + mock_cache.get_cached_result.return_value = cached_result + + result2 = simulator.run_simulation(test_params) + + # Verify simulation was NOT run again + simulator.server.run_sim_with_datastream.assert_not_called() + mock_cache.cache_result.assert_not_called() + + # Verify we got cached result + assert result2 == cached_result, "Did not return cached result" + + def test_cache_different_parameters_trigger_new_simulation(self, simulator): + """Test that different parameters trigger new simulations.""" + # Mock PLECS connection + simulator.server = MagicMock() + simulator.plecs_app = MagicMock() + + # Mock cache behavior + mock_cache = MagicMock() + simulator.cache = mock_cache + mock_cache.get_cached_result.return_value = None # Always cache miss + + mock_result = { + 'Time': [0, 1e-6, 2e-6], + 'Values': [[1, 2, 3], [4, 5, 6]] + } + simulator.server.run_sim_with_datastream.return_value = mock_result + + # First simulation + params1 = {'Vi': 24.0, 'Lo': 15e-6} + result1 = simulator.run_simulation(params1) + + # Second simulation with different parameters + params2 = {'Vi': 24.0, 'Lo': 25e-6} # Different Lo value + result2 = simulator.run_simulation(params2) + + # Verify both simulations were run + assert simulator.server.run_sim_with_datastream.call_count == 2 + assert mock_cache.cache_result.call_count == 2 + + # Verify different parameters were set + assert simulator.server.load_modelvars.call_count == 2 + + # Get the parameters for each call + call1_params = simulator.server.load_modelvars.call_args_list[0][0][0] + call2_params = simulator.server.load_modelvars.call_args_list[1][0][0] + + assert call1_params['Lo'] != call2_params['Lo'], "Parameters should be different" + + def test_plecs_file_change_triggers_new_simulation(self, temp_cache_dir): + """Test that changing PLECS file triggers new simulation even with same parameters.""" + # Create test PLECS files + original_file = temp_cache_dir / "test1.plecs" + modified_file = temp_cache_dir / "test2.plecs" + + # Create simple PLECS content + plecs_content1 = """ + Plecs { + Name "test1" + InitializationCommands "Vi = 24;" + } + """ + plecs_content2 = """ + Plecs { + Name "test2" + InitializationCommands "Vi = 24; Lo = 15e-6;" + } + """ + + original_file.write_text(plecs_content1) + modified_file.write_text(plecs_content2) + + # Create real cache instance + cache = SimulationCache() + + # Same parameters, different files + params = {'Vi': 24.0, '_simulation_type': 'real_plecs'} + + # Simulate results for both files + timeseries1 = pd.DataFrame({'Time': [1, 2], 'Signal': [0.1, 0.2]}) + metadata1 = {'file': str(original_file)} + timeseries2 = pd.DataFrame({'Time': [3, 4], 'Signal': [0.3, 0.4]}) + metadata2 = {'file': str(modified_file)} + + # Cache result for first file + cache.cache_result(str(original_file), params, timeseries1, metadata1) + + # Try to get cached result for second file (should be None - cache miss) + cached_result = cache.get_cached_result(str(modified_file), params) + assert cached_result is None, "Different PLECS files should not share cache" + + # Cache result for second file + cache.cache_result(str(modified_file), params, timeseries2, metadata2) + + # Verify both files have separate cache entries + cached1 = cache.get_cached_result(str(original_file), params) + cached2 = cache.get_cached_result(str(modified_file), params) + + assert cached1 is not None, "Original file cache missing" + assert cached2 is not None, "Modified file cache missing" + # Compare metadata to verify they're different cache entries + assert cached1['metadata'] != cached2['metadata'], ( + "Different files should have different cached results" + ) + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/tests/test_real_plecs_integration.py b/tests/integration/test_real_plecs_integration.py similarity index 63% rename from tests/test_real_plecs_integration.py rename to tests/integration/test_real_plecs_integration.py index cb88d75..382c076 100644 --- a/tests/test_real_plecs_integration.py +++ b/tests/integration/test_real_plecs_integration.py @@ -197,7 +197,7 @@ def test_parameter_sweep_generation(self, model_file): # Verify Co values are in range assert 100e-6 <= point['Co'] <= 500e-6, f"Co value {point['Co']} out of range" - + def test_cache_identical_simulations(self, simulator): """Test that identical simulations use cache instead of re-running.""" # Mock PLECS connection @@ -342,180 +342,7 @@ def test_plecs_file_change_triggers_new_simulation(self, temp_cache_dir): assert cached1['metadata'] != cached2['metadata'], ( "Different files should have different cached results" ) - - def test_expression_variables_are_skipped(self, simulator): - """Test that expression variables (like D=Vo_ref/Vi) are skipped during parameter setting.""" - # Mock PLECS connection - simulator.server = MagicMock() - simulator.plecs_app = MagicMock() - - # Parameters with expressions (should be skipped) and simple values (should be set) - test_params = { - 'Vi': 24.0, # Should be set - 'Lo': 15e-6, # Should be set - 'D': 'Vo_ref/Vi', # Should be skipped (expression) - 'Po': 'Vi*Ii_max', # Should be skipped (expression) - 'Ro': 'Vo_ref^2/Po', # Should be skipped (expression) - 'fs': 100000.0 # Should be set - } - - # Call parameter setting - simulator.set_parameters(test_params) - - # Verify load_modelvars was called - simulator.server.load_modelvars.assert_called_once() - - # Get the actual parameters passed to PLECS - called_params = simulator.server.load_modelvars.call_args[0][0] - - # Verify simple parameters were included - assert 'Vi' in called_params, "Vi should be set" - assert 'Lo' in called_params, "Lo should be set" - assert 'fs' in called_params, "fs should be set" - - # Verify expression parameters were skipped - assert 'D' not in called_params, "Expression D should be skipped" - assert 'Po' not in called_params, "Expression Po should be skipped" - assert 'Ro' not in called_params, "Expression Ro should be skipped" - - # Verify correct values - assert called_params['Vi'] == 24.0 - assert called_params['Lo'] == 15e-6 - assert called_params['fs'] == 100000.0 - - def test_end_to_end_parameter_sweep(self, model_file): - """Integration test: full parameter sweep with cache verification.""" - # This test requires PLECS to be installed and working - simulator = RealPlecsSimulator(model_file) - - try: - # Start PLECS (skip if not available) - if not simulator.start_plecs_and_connect(): - pytest.skip("PLECS not available for integration test") - - # Parse model - parsed_data = parse_plecs_file(model_file) - base_params = parsed_data['init_vars'] - - # Create small parameter sweep - sim_plan = SimulationPlan(model_file, base_params) - sim_plan.add_sweep_parameter('Lo', 10e-6, 20e-6, 2) # Just 2 points - - simulation_points = sim_plan.generate_simulation_points() - assert len(simulation_points) == 2, "Should generate 2 simulation points" - - # Run simulations - results = [] - for params in simulation_points: - result = simulator.run_simulation(params) - assert result['metadata']['success'], "Simulation should succeed" - results.append(result) - - # Verify results are different (different Lo values should give different results) - ts1 = results[0]['timeseries'] - ts2 = results[1]['timeseries'] - - # Results should have same structure but potentially different values - if hasattr(ts1, 'columns'): # pandas DataFrame - assert list(ts1.columns) == list(ts2.columns), "Column structure should match" - assert len(ts1) == len(ts2), "Should have same number of time points" - else: # dict format - assert ts1.keys() == ts2.keys(), "Keys should match" - assert len(ts1['Time']) == len(ts2['Time']), "Should have same time points" - - # Test cache behavior - run same simulation again - cached_result = simulator.run_simulation(simulation_points[0]) - - # Should get same result (either from cache or fresh simulation) - assert cached_result['metadata']['success'], "Cached/repeated simulation should succeed" - - finally: - simulator.close() - - -class TestSimulationCache: - """Specific tests for simulation cache behavior.""" - - @pytest.fixture - def temp_cache_dir(self): - """Temporary cache directory.""" - temp_dir = tempfile.mkdtemp() - yield Path(temp_dir) - shutil.rmtree(temp_dir) - - def test_cache_key_includes_file_hash(self, temp_cache_dir): - """Test that cache keys include file hash to detect file changes.""" - cache = SimulationCache() - - # Create test files with different content - file1 = temp_cache_dir / "test1.plecs" - file2 = temp_cache_dir / "test2.plecs" - - file1.write_text("Plecs { Name 'test1' }") - file2.write_text("Plecs { Name 'test2' }") - - params = {'Vi': 24.0} - timeseries = pd.DataFrame({'Time': [1, 2], 'Signal': [0.1, 0.2]}) - metadata = {'test': 'data'} - - # Cache results for both files - cache.cache_result(str(file1), params, timeseries, metadata) - cache.cache_result(str(file2), params, timeseries, metadata) - - # Verify separate cache entries exist - cached1 = cache.get_cached_result(str(file1), params) - cached2 = cache.get_cached_result(str(file2), params) - - assert cached1 is not None, "File1 should be cached" - assert cached2 is not None, "File2 should be cached" - - # Modify file1 content - file1.write_text("Plecs { Name 'test1_modified' }") - - # Cache should miss for modified file1 - cached1_after = cache.get_cached_result(str(file1), params) - assert cached1_after is None, "Modified file should cause cache miss" - - # Cache should still hit for unchanged file2 - cached2_after = cache.get_cached_result(str(file2), params) - assert cached2_after is not None, "Unchanged file should still be cached" - - def test_cache_isolation_by_simulation_type(self, temp_cache_dir): - """Test that different simulation types don't share cache.""" - cache = SimulationCache() - - file_path = str(temp_cache_dir / "test.plecs") - Path(file_path).write_text("Plecs { Name 'test' }") - - # Same parameters but different simulation types - params_mock = {'Vi': 24.0, '_simulation_type': 'mock'} - params_real = {'Vi': 24.0, '_simulation_type': 'real_plecs'} - - timeseries_mock = pd.DataFrame({'Time': [1, 2], 'Signal': [0.1, 0.2]}) - metadata_mock = {'simulation_type': 'mock'} - timeseries_real = pd.DataFrame({'Time': [1, 2], 'Signal': [0.3, 0.4]}) - metadata_real = {'simulation_type': 'real_plecs'} - - # Cache both results - cache.cache_result( - file_path, params_mock, timeseries_mock, metadata_mock - ) - cache.cache_result( - file_path, params_real, timeseries_real, metadata_real - ) - - # Verify separate cache entries - cached_mock = cache.get_cached_result(file_path, params_mock) - cached_real = cache.get_cached_result(file_path, params_real) - - assert cached_mock is not None, "Mock simulation should be cached" - assert cached_real is not None, "Real simulation should be cached" - # Compare metadata to verify they're different cache entries - assert cached_mock['metadata'] != cached_real['metadata'], ( - "Different simulation types should have different cache" - ) if __name__ == "__main__": - # Run specific test - pytest.main([__file__, "-v", "--tb=short"]) + pytest.main([__file__, "-v"]) diff --git a/tests/test_simulate_batch.py b/tests/integration/test_simulate_batch.py similarity index 100% rename from tests/test_simulate_batch.py rename to tests/integration/test_simulate_batch.py diff --git a/tests/integration/test_simulation.py b/tests/integration/test_simulation.py new file mode 100644 index 0000000..5d6a58a --- /dev/null +++ b/tests/integration/test_simulation.py @@ -0,0 +1,49 @@ +import unittest +from pathlib import Path +from pyplecs import PlecsServer + + +class SimulationTestSuite(unittest.TestCase): + """Test cases for simulation management.""" + + def test_run_sim_single(self): + """Test single simulation execution.""" + sim_file_path = Path('../data/simple_buck.plecs') + try: + plecs_server = PlecsServer( + sim_path=str(sim_file_path.parent), sim_name=sim_file_path.name + ) + except Exception as exc: + import pytest + + pytest.skip(f"PLECS RPC not available: {exc}") + + inputs = {'Vi': 250, 'Ii_max': 25, 'Vo_ref': 25} + results = plecs_server.run_sim_single(inputs) + + self.assertIsNotNone(results) + self.assertTrue(hasattr(results, 'Time')) + self.assertTrue(hasattr(results, 'Values')) + + def test_run_sim_with_datastream(self): + """Test simulation with datastream parameters.""" + sim_file_path = Path('../data/simple_buck.plecs') + try: + plecs_server = PlecsServer( + sim_path=str(sim_file_path.parent), sim_name=sim_file_path.name + ) + except Exception as exc: + import pytest + + pytest.skip(f"PLECS RPC not available: {exc}") + + params = {'Vi': 25, 'Ii_max': 1, 'Vo_ref': 5} + results = plecs_server.run_sim_with_datastream(param_dict=params) + + self.assertIsNotNone(results) + self.assertTrue(hasattr(results, 'Time')) + self.assertTrue(hasattr(results, 'Values')) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_interactive.py b/tests/interactive/test_interactive.py similarity index 93% rename from tests/test_interactive.py rename to tests/interactive/test_interactive.py index 889dc61..55eb321 100644 --- a/tests/test_interactive.py +++ b/tests/interactive/test_interactive.py @@ -1,12 +1,23 @@ # -*- coding: utf-8 -*- - +# Warn or fail if not run with pytest -s (no output capture) +import os import sys + + import unittest import time from pathlib import Path +if os.environ.get("PYTEST_CURRENT_TEST") and not sys.stdout.isatty(): + raise RuntimeError( + "This test requires pytest -s (no output capture) for input() to work.\n" + "Run with: pytest -s tests/interactive/test_interactive.py" + ) + + # Add project root to path for import -sys.path.insert(0, str(Path(__file__).parent.parent)) +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + import pyplecs @@ -25,10 +36,9 @@ def test01_sequential_simulation_server_same_file(self): full_sim_name = str(sim_file_path_obj.absolute()) plecs_mdl = pyplecs.GenericConverterPlecsMdl(full_sim_name) + print("\n=== Starting Sequential Simulation Test ===") if not sys.stdin.isatty(): self.skipTest('Interactive tests require a TTY; skipping in CI') - - print("\n=== Starting Sequential Simulation Test ===") input("Press Enter to continue with next sim 00...") # try to reach PLECS RPC and skip if unavailable @@ -160,4 +170,4 @@ def test02_sequential_simulation_server_different_file(self): print("\nOr run all interactive tests:") print("python -m pytest tests/test_interactive.py -v -s") - unittest.main() + unittest.main() \ No newline at end of file diff --git a/tests/test_basic.py b/tests/test_basic.py deleted file mode 100644 index 9678ef6..0000000 --- a/tests/test_basic.py +++ /dev/null @@ -1,37 +0,0 @@ -# -*- coding: utf-8 -*- -""" -DEPRECATED: This file is being reorganized into separate test files. - -New test file organization: -- test_automated.py: Tests that run without user interaction -- test_interactive.py: Tests that require manual user input -- test_gui_automation.py: Tests that require GUI automation with external apps - -Please use the appropriate test file for your testing needs: - -For automated CI/CD testing: - python -m pytest tests/test_automated.py -v - -For interactive testing: - python -m pytest tests/test_interactive.py -v -s - -For GUI automation testing: - python -m pytest tests/test_gui_automation.py -v -s - -For backward compatibility, this file redirects to test_automated.py -""" - -from .test_automated import AutomatedTestSuite - -# Re-export the automated test suite for backward compatibility -BasicTestSuite = AutomatedTestSuite - -if __name__ == '__main__': - import unittest - print("NOTICE: test_basic.py is deprecated.") - print("Please use the new organized test files:") - print("- tests/test_automated.py") - print("- tests/test_interactive.py") - print("- tests/test_gui_automation.py") - print("\nRunning automated tests for backward compatibility...") - unittest.main() diff --git a/tests/test_cache_behavior.py b/tests/test_cache_behavior.py deleted file mode 100644 index 3355db8..0000000 --- a/tests/test_cache_behavior.py +++ /dev/null @@ -1,225 +0,0 @@ -""" -Tests for cache behavior with file changes and simulation type isolation. -""" - -# Move imports to the top -import pytest -import tempfile -import shutil -from pathlib import Path -from pyplecs.cache import SimulationCache -from cli_demo_nomocks import RealPlecsSimulator - - -class TestCacheBehavior: - """Tests for simulation cache behavior.""" - - @pytest.fixture - def temp_dir(self): - """Create temporary directory for test files.""" - temp_dir = tempfile.mkdtemp() - yield Path(temp_dir) - shutil.rmtree(temp_dir) - - @pytest.fixture - def cache(self): - """Create simulation cache instance.""" - return SimulationCache() - - def test_cache_differentiates_simulation_types(self, cache, temp_dir): - """Test that mock and real simulations don't share cache.""" - # Create test PLECS file - test_file = temp_dir / "test.plecs" - test_file.write_text( - "Plecs { Name 'test' InitializationCommands 'Vi=24;' }" - ) - - # Same base parameters but different simulation types - base_params = {'Vi': 24.0, 'Lo': 15e-6} - - mock_params = base_params.copy() - mock_params['_simulation_type'] = 'mock' # type: ignore - mock_params['_simulation_engine'] = 'mock' # type: ignore - - real_params = base_params.copy() - real_params['_simulation_type'] = 'real_plecs' # type: ignore - real_params['_simulation_engine'] = 'xml_rpc' # type: ignore - - # Create mock result structure - mock_result_data = {'Time': [0, 1, 2], 'Signal_0': [1, 2, 3]} - mock_metadata = {'simulation_type': 'mock', 'success': True} - - # Cache mock result - cache.cache_result( - str(test_file), - mock_params, - mock_result_data, - mock_metadata - ) - - # Check that real simulation parameters don't hit mock cache - cached_result = cache.get_cached_result(str(test_file), real_params) - assert cached_result is None, "Real params should not hit mock cache" - - # Create and cache real result - real_result_data = {'Time': [0, 1, 2], 'Signal_0': [4, 5, 6]} - real_metadata = {'simulation_type': 'real_plecs', 'success': True} - - cache.cache_result( - str(test_file), - real_params, - real_result_data, - real_metadata - ) - - # Verify both types have separate cache entries - mock_cached = cache.get_cached_result(str(test_file), mock_params) - real_cached = cache.get_cached_result(str(test_file), real_params) - - assert mock_cached is not None, "Mock result should be cached" - assert real_cached is not None, "Real result should be cached" - # Compare metadata to verify they're different cache entries - assert mock_cached['metadata'] != real_cached['metadata'], ( - "Should have different cached results" - ) - - def test_cache_detects_file_changes(self, cache, temp_dir): - """Test that cache misses when PLECS file content changes.""" - test_file = temp_dir / "test.plecs" - - # Create initial file - initial_content = ( - "Plecs { Name 'test1' InitializationCommands 'Vi=24;' }" - ) - test_file.write_text(initial_content) - - params = { - 'Vi': 24.0, - '_simulation_type': 'real_plecs', - '_simulation_engine': 'xml_rpc' - } - - # Cache initial result - initial_result = {'Time': [0, 1], 'Signal_0': [1, 2]} - initial_metadata = {'success': True, 'file_content': initial_content} - - cache.cache_result( - str(test_file), params, initial_result, initial_metadata - ) - - # Verify cache hit - cached = cache.get_cached_result(str(test_file), params) - assert cached is not None, ( - "Should find cached result for original file" - ) - - # Modify file content - modified_content = ( - "Plecs { Name 'test2' InitializationCommands 'Vi=48;' }" - ) - test_file.write_text(modified_content) - - # Should get cache miss for modified file - cached_after_change = cache.get_cached_result(str(test_file), params) - assert cached_after_change is None, ( - "Should miss cache after file change" - ) - - def test_cache_performance_with_real_simulator(self): - """Test cache performance with real PLECS simulator.""" - model_file = Path("data/simple_buck.plecs") - if not model_file.exists(): - pytest.skip("PLECS model file not found") - - simulator = RealPlecsSimulator(str(model_file)) - success = simulator.start_plecs_and_connect() - if not success: - simulator.close() - pytest.skip("PLECS not available") - - try: - params = { - 'Vi': 24.0, - 'Lo': 15e-6, - '_simulation_type': 'real_plecs', - '_simulation_engine': 'xml_rpc' - } - - # First run - should execute simulation - import time - start = time.time() - result1 = simulator.run_simulation(params) - time1 = time.time() - start - - assert result1['metadata']['success'], "First simulation failed" - - # Second run - should use cache - start = time.time() - result2 = simulator.run_simulation(params) - time2 = time.time() - start - - assert result2['metadata']['success'], "Second simulation failed" - - # Cache should provide significant speedup - speedup = time1 / time2 if time2 > 0 else float('inf') - assert speedup > 1.5, ( - f"Cache not effective: {time1:.3f}s vs {time2:.3f}s" - ) - - finally: - simulator.close() - - def test_cache_format_compatibility(self, cache, temp_dir): - """Test compatibility between DataFrame and configured formats.""" - import pandas as pd - import pyarrow as pa - import pyarrow.parquet as pq - import json - - # Create mock DataFrame - df = pd.DataFrame({ - 'Time': [0, 1, 2], - 'Signal_0': [1.0, 2.0, 3.0] - }) - - # Convert DataFrame to parquet - parquet_file = temp_dir / "timeseries.parquet" - table = pa.Table.from_pandas(df) - with parquet_file.open('wb') as f: - pq.write_table(table, f) - - # Read back parquet file - read_table = pq.read_table(parquet_file) - read_df = read_table.to_pandas() - assert df.equals(read_df), "Parquet format mismatch" - - # Convert metadata to JSON - metadata = {'simulation_type': 'real_plecs', 'success': True} - json_file = temp_dir / "metadata.json" - with json_file.open('w', encoding='utf-8') as f: - json.dump(metadata, f) - - # Read back JSON file - with json_file.open('r', encoding='utf-8') as f: - read_metadata = json.load(f) - assert metadata == read_metadata, "JSON format mismatch" - - # Cache and retrieve using both formats - params = { - 'Vi': 24.0, - '_simulation_type': 'real_plecs', - '_simulation_engine': 'xml_rpc' - } - cache.cache_result(str(parquet_file), params, read_df, read_metadata) - cached_result = cache.get_cached_result(str(parquet_file), params) - assert cached_result is not None, "Cached result not found" - assert cached_result['timeseries'].equals(read_df), ( - "Cached timeseries mismatch" - ) - assert cached_result['metadata'] == read_metadata, ( - "Cached metadata mismatch" - ) - - -if __name__ == "__main__": - pytest.main([__file__, "-v"]) diff --git a/tests/test_entrypoint.py b/tests/test_entrypoint.py deleted file mode 100644 index 6e30630..0000000 --- a/tests/test_entrypoint.py +++ /dev/null @@ -1,17 +0,0 @@ -import subprocess -import sys - - -def test_entrypoint_help(): - # Call the installed console script and check it returns 0 and prints help - cmd = [sys.executable.replace('python', 'pyplecs-setup') if 'python' in sys.executable else 'pyplecs-setup', '--help'] - # Fallback to direct module call - try: - res = subprocess.run(['pyplecs-setup', '--help'], capture_output=True, text=True) - assert res.returncode == 0 - assert 'PyPLECS setup helper' in res.stdout - except FileNotFoundError: - # fallback: call as module - res = subprocess.run([sys.executable, '-m', 'pyplecs.cli.installer', '--help'], capture_output=True, text=True) - assert res.returncode == 0 - assert 'PyPLECS setup helper' in res.stdout diff --git a/tests/test_gui_automation.py b/tests/test_gui_automation.py deleted file mode 100644 index 2def37b..0000000 --- a/tests/test_gui_automation.py +++ /dev/null @@ -1,111 +0,0 @@ -# -*- coding: utf-8 -*- - -import sys -import unittest -import time -from pathlib import Path - -# Add project root to path for import -sys.path.insert(0, str(Path(__file__).parent.parent)) - -import pyplecs -from pywinauto.application import Application - - -class GUIAutomationTestSuite(unittest.TestCase): - """GUI automation test cases that require external application interaction. - - These tests use pywinauto to interact with GUI applications and may - fail in headless environments or when the required applications are - not available. - """ - - def test01_pywinauto_with_notepad(self): - """Test pywinauto functionality with Windows Notepad. - - This test verifies that pywinauto can interact with external - Windows applications by automating Notepad operations. - """ - # Run a target application - # Skip GUI tests on non-Windows or headless CI environments - import sys - - if sys.platform != 'win32': - self.skipTest('pywinauto tests require Windows') - - app = Application().start("notepad.exe") - # Select a menu item - app.UntitledNotepad.menu_select("Help->About Notepad") - # Click on a button - app.AboutNotepad.OK.click() - # Type a text string - app.UntitledNotepad.Edit.type_keys("pywinauto Works!", - with_spaces=True) - - def test02_gui_simulation(self): - """Test PLECS GUI automation for simulation execution. - - This test demonstrates automated interaction with the PLECS GUI - to load models and run simulations through the graphical interface. - """ - sim_file_path_obj = Path('data/simple_buck.plecs') - full_sim_name = str(sim_file_path_obj.absolute()) - buck_mdl = pyplecs.GenericConverterPlecsMdl(full_sim_name) - - # Create simulation variants - ModelVars = dict() - ModelVars["Vi"] = 250 - ModelVars["Ii_max"] = 25 - ModelVars["Vo_ref"] = 25 - - buck_mdl_01 = pyplecs.generate_variant_plecs_mdl( - src_mdl=buck_mdl, variant_name='01', variant_vars=ModelVars) - - ModelVars["Vi"] = 25 - ModelVars["Ii_max"] = 1 - ModelVars["Vo_ref"] = 5 - - buck_mdl_02 = pyplecs.generate_variant_plecs_mdl( - src_mdl=buck_mdl, variant_name='02', variant_vars=ModelVars) - - # Initialize PLECS servers - time.sleep(1) - buck_mdl_01_server = pyplecs.PlecsServer(buck_mdl_01.folder, - buck_mdl_01.simulation_name) - - time.sleep(1) - buck_mdl_02_server = pyplecs.PlecsServer(buck_mdl_02.folder, - buck_mdl_02.simulation_name) - - # Start PLECS application - # Skip PLECS GUI automation if RPC server is not reachable - import socket - - try: - s = socket.create_connection(('localhost', 1080), timeout=1) - s.close() - except Exception: - self.skipTest('PLECS RPC not reachable; skipping GUI automation') - - plecs42 = pyplecs.PlecsApp() - plecs42.open_plecs() - - # Run simulations via GUI - buck_mdl_01_server.load() - time.sleep(1) - plecs42.run_simulation_by_gui(plecs_mdl=buck_mdl_01) - - buck_mdl_02_server.load() - time.sleep(1) - plecs42.run_simulation_by_gui(plecs_mdl=buck_mdl_02) - - -if __name__ == '__main__': - print("WARNING: This test suite contains GUI automation tests that require") - print("external applications and may fail in headless environments.") - print("\nThese tests should typically be run manually or in environments") - print("with full GUI support and the required applications installed.") - print("\nExample usage:") - print("python -m pytest tests/test_gui_automation.py -v -s") - - unittest.main() diff --git a/tests/test_parser.py b/tests/test_parser.py deleted file mode 100644 index 6089d1b..0000000 --- a/tests/test_parser.py +++ /dev/null @@ -1,49 +0,0 @@ -import unittest -from pathlib import Path -from pyplecs.plecs_parser import ( - parse_plecs_file, - plecs_overview, - scan_plecs_dir -) - - -class ParserTestSuite(unittest.TestCase): - """Test cases for PLECS file parser.""" - - def test_parse_plecs_file(self): - """Test parsing a single .plecs file.""" - test_file = Path('data/simple_buck.plecs') - result = parse_plecs_file(str(test_file)) - - self.assertIsInstance(result, dict) - self.assertIn('file', result) - self.assertIn('components', result) - self.assertIn('init_vars', result) - self.assertGreater(len(result['components']), 0) - - def test_plecs_overview(self): - """Test generating an overview for a .plecs file.""" - test_file = Path('data/simple_buck.plecs') - overview = plecs_overview(str(test_file)) - - self.assertIsInstance(overview, dict) - self.assertIn('file', overview) - self.assertIn('components', overview) - self.assertIn('init_vars', overview) - self.assertGreater(len(overview['components']), 0) - - def test_scan_plecs_dir(self): - """Test scanning a directory for .plecs files.""" - test_dir = Path('data') - results = scan_plecs_dir(str(test_dir)) - - self.assertIsInstance(results, dict) - self.assertGreater(len(results), 0) - for _, result in results.items(): - self.assertIn('file', result) - self.assertIn('components', result) - self.assertIn('init_vars', result) - - -if __name__ == '__main__': - unittest.main() diff --git a/tests/test_parser_and_core.py b/tests/test_parser_and_core.py deleted file mode 100644 index e69de29..0000000 diff --git a/tests/test_plecs_integration_simple.py b/tests/test_plecs_integration_simple.py deleted file mode 100644 index 8b4b860..0000000 --- a/tests/test_plecs_integration_simple.py +++ /dev/null @@ -1,137 +0,0 @@ -""" -Simple test to verify real PLECS integration works end-to-end. -""" - -import pytest -from pathlib import Path -import tempfile -import shutil - -# Import our modules -from pyplecs.plecs_parser import parse_plecs_file - -# Import the CLI components -import sys -import os -sys.path.append(os.path.dirname(os.path.dirname(__file__))) -from cli_demo_nomocks import RealPlecsSimulator, SimulationPlan - - -def test_real_plecs_parameter_sweep(): - """ - Test that real PLECS simulations work with parameter sweeps. - This is the most important test - it verifies: - 1. PLECS starts and connects via XML-RPC - 2. Parameters are set correctly - 3. Simulations execute and return real data - 4. Parameter sweeps work - 5. Cache behavior is correct - """ - model_file = Path("data/simple_buck.plecs") - - if not model_file.exists(): - pytest.skip("PLECS model file not found") - - # Test 1: Parse model file - parsed_data = parse_plecs_file(model_file) - assert len(parsed_data['init_vars']) > 0, "Should find initialization variables" - - # Test 2: Create simulator and start PLECS - simulator = RealPlecsSimulator(model_file) - - try: - success = simulator.start_plecs_and_connect() - if not success: - pytest.skip("PLECS not available or failed to start") - - # Test 3: Verify we can connect - assert simulator.server is not None, "XML-RPC server should be initialized" - assert simulator.plecs_app is not None, "PLECS app should be initialized" - - # Test 4: Run a simple simulation - simple_params = { - 'Vi': 24.0, - 'Lo': 15e-6, - 'Co': 150e-6, - '_simulation_type': 'real_plecs', - '_simulation_engine': 'xml_rpc' - } - - result1 = simulator.run_simulation(simple_params) - - # Verify result structure - assert 'timeseries' in result1, "Should have timeseries data" - assert 'metadata' in result1, "Should have metadata" - assert result1['metadata']['success'], "Simulation should succeed" - - # Verify we have actual time series data - timeseries = result1['timeseries'] - if hasattr(timeseries, 'columns'): # pandas DataFrame - assert 'Time' in timeseries.columns, "Should have Time column" - assert len(timeseries) > 100, "Should have substantial data points" - else: # dict format - assert 'Time' in timeseries, "Should have Time data" - assert len(timeseries['Time']) > 100, "Should have substantial time points" - - print(f"✓ First simulation successful with {len(timeseries)} data points") - - # Test 5: Run simulation with different parameters - different_params = simple_params.copy() - different_params['Lo'] = 25e-6 # Different inductance - - result2 = simulator.run_simulation(different_params) - assert result2['metadata']['success'], "Second simulation should succeed" - - print("✓ Second simulation with different parameters successful") - - # Test 6: Run parameter sweep - base_params = parsed_data['init_vars'].copy() - sim_plan = SimulationPlan(model_file, base_params) - - # Small sweep: 2 points for Lo - sim_plan.add_sweep_parameter('Lo', 10e-6, 20e-6, 2) - - simulation_points = sim_plan.generate_simulation_points() - assert len(simulation_points) == 2, "Should generate 2 simulation points" - - # Run all simulations in sweep - sweep_results = [] - for i, params in enumerate(simulation_points): - print(f"Running sweep simulation {i+1}/2...") - result = simulator.run_simulation(params) - assert result['metadata']['success'], f"Sweep simulation {i+1} should succeed" - sweep_results.append(result) - - print("✓ Parameter sweep completed successfully") - - # Test 7: Verify results are different (different Lo should give different results) - ts1 = sweep_results[0]['timeseries'] - ts2 = sweep_results[1]['timeseries'] - - # Results should have same structure - if hasattr(ts1, 'columns'): # pandas DataFrame - assert list(ts1.columns) == list(ts2.columns), "Column structure should match" - assert len(ts1) == len(ts2), "Should have same number of time points" - else: # dict format - assert ts1.keys() == ts2.keys(), "Keys should match" - assert len(ts1['Time']) == len(ts2['Time']), "Should have same time points" - - print("✓ Sweep results have consistent structure") - - # Test 8: Test cache behavior - run identical simulation again - print("Testing cache behavior...") - - # Run same simulation as first one - cached_result = simulator.run_simulation(simple_params) - assert cached_result['metadata']['success'], "Cached simulation should succeed" - - print("✓ Cache test completed") - - print("🎉 ALL TESTS PASSED - Real PLECS integration is working!") - - finally: - simulator.close() - - -if __name__ == "__main__": - test_real_plecs_parameter_sweep() diff --git a/tests/test_simulation.py b/tests/test_simulation.py deleted file mode 100644 index 2afa5a3..0000000 --- a/tests/test_simulation.py +++ /dev/null @@ -1,39 +0,0 @@ -import unittest -from pathlib import Path -from pyplecs import PlecsServer - - -class SimulationTestSuite(unittest.TestCase): - """Test cases for simulation management.""" - - def test_run_sim_single(self): - """Test single simulation execution.""" - sim_file_path = Path('../data/simple_buck.plecs') - plecs_server = PlecsServer( - sim_path=str(sim_file_path.parent), sim_name=sim_file_path.name - ) - - inputs = {'Vi': 250, 'Ii_max': 25, 'Vo_ref': 25} - results = plecs_server.run_sim_single(inputs) - - self.assertIsNotNone(results) - self.assertTrue(hasattr(results, 'Time')) - self.assertTrue(hasattr(results, 'Values')) - - def test_run_sim_with_datastream(self): - """Test simulation with datastream parameters.""" - sim_file_path = Path('../data/simple_buck.plecs') - plecs_server = PlecsServer( - sim_path=str(sim_file_path.parent), sim_name=sim_file_path.name - ) - - params = {'Vi': 25, 'Ii_max': 1, 'Vo_ref': 5} - results = plecs_server.run_sim_with_datastream(param_dict=params) - - self.assertIsNotNone(results) - self.assertTrue(hasattr(results, 'Time')) - self.assertTrue(hasattr(results, 'Values')) - - -if __name__ == '__main__': - unittest.main() diff --git a/tests/test_smoke.py b/tests/test_smoke.py deleted file mode 100644 index 4b52135..0000000 --- a/tests/test_smoke.py +++ /dev/null @@ -1,97 +0,0 @@ -#!/usr/bin/env python3 -""" -Lightweight smoke test for PyPLECS installation validation. - -This test checks that the core modules can be imported and basic functionality -works without requiring GUI automation or external PLECS installation. -""" - -import sys -import unittest -from pathlib import Path - -# Add project root to path for import -sys.path.insert(0, str(Path(__file__).parent.parent)) - -class SmokeTestSuite(unittest.TestCase): - """Lightweight tests to validate PyPLECS installation.""" - - def test01_basic_imports(self): - """Test that core modules can be imported.""" - try: - import pyplecs - import pyplecs.config - import pyplecs.core.models - # Test that lazy imports don't fail immediately - from pyplecs.cache import SimulationCache - from pyplecs.orchestration import SimulationOrchestrator - except ImportError as e: - self.fail(f"Failed to import core modules: {e}") - - def test02_config_loading(self): - """Test that configuration can be loaded.""" - try: - from pyplecs.config import get_config - config = get_config() - # Basic validation - config should be a dict-like object - self.assertIsNotNone(config) - except Exception as e: - self.fail(f"Failed to load configuration: {e}") - - def test03_models_import(self): - """Test that core models can be imported.""" - try: - from pyplecs.core.models import ( - SimulationRequest, SimulationResult, SimulationStatus - ) - - # Test that classes are available - self.assertTrue(callable(SimulationRequest)) - self.assertTrue(callable(SimulationResult)) - - # Test that enum values are accessible - self.assertEqual(SimulationStatus.COMPLETED.value, "completed") - self.assertEqual(SimulationStatus.FAILED.value, "failed") - - except Exception as e: - self.fail(f"Failed to import core models: {e}") - - def test04_webgui_import(self): - """Test that web GUI components can be imported.""" - try: - from pyplecs.webgui import create_web_app - # Don't actually create the app to avoid dependency issues - self.assertTrue(callable(create_web_app)) - except ImportError as e: - self.fail(f"Failed to import web GUI components: {e}") - - def test05_cli_installer_import(self): - """Test that CLI installer can be imported and has main function.""" - try: - from pyplecs.cli.installer import main - self.assertTrue(callable(main)) - except ImportError as e: - self.fail(f"Failed to import CLI installer: {e}") - - def test06_package_version(self): - """Test that package version is accessible.""" - try: - import pyplecs - # Basic check that version info exists - self.assertTrue(hasattr(pyplecs, '__version__') or - hasattr(pyplecs, 'VERSION') or - 'version' in dir(pyplecs)) - except Exception as e: - # Version info is optional, so just warn - print(f"Warning: Could not access version info: {e}") - -if __name__ == '__main__': - print("Running PyPLECS installation smoke tests...") - print("=" * 60) - - # Run tests with verbose output - unittest.main(verbosity=2, exit=False) - - print("\n" + "=" * 60) - print("Smoke tests completed. If all tests passed, PyPLECS is properly installed.") - print("For full functionality testing, ensure PLECS is installed and run test_basic.py") diff --git a/tests/unit/test_automated.py b/tests/unit/test_automated.py deleted file mode 100644 index e69de29..0000000 diff --git a/tools/installers/windows_installer.ps1 b/tools/installers/windows_installer.ps1 index 5a5f224..01a7de1 100644 --- a/tools/installers/windows_installer.ps1 +++ b/tools/installers/windows_installer.ps1 @@ -172,7 +172,7 @@ if (Test-Path $activate) { $venvPython = Join-Path $venvPath 'Scripts\python.exe' try { & $venvPython -m pip install --upgrade pip - & $venvPython -m pip install fastapi uvicorn[standard] jinja2 pandas pyyaml pywin32 pywinauto --progress-bar off + & $venvPython -m pip install fastapi uvicorn[standard] jinja2 pandas pyyaml pywin32 --progress-bar off if ($LASTEXITCODE -ne 0) { Log "pip install inside venv failed with exit code $LASTEXITCODE"; $global:ExitCode = 5; $global:StatusMessage = 'pip_install_failed'; Write-StatusJson $global:ExitCode $global:StatusMessage; exit $global:ExitCode } } catch { Log "Exception during pip install in venv: $_"