From fe96222e9dc201bd721c4eb5beaafe3b054ffded Mon Sep 17 00:00:00 2001 From: netanelC Date: Wed, 6 May 2026 23:37:11 +0300 Subject: [PATCH 1/2] refactor: some gaps --- Ammeters/base_ammeter.py | 24 ++++++++++-- README.md | 58 +++++++++++++++++++++++++--- config/config.yaml | 1 + examples/compare_runs.py | 60 ++++++++++++++++++++-------- examples/run_framework.py | 6 +-- main.py | 28 ++++++++------ src/testing/test_framework.py | 61 ++++++++++++++--------------- src/utils/analysis.py | 16 ++++++-- src/utils/comparison.py | 73 +++++++++++++++++------------------ src/utils/logger.py | 22 ++++++++--- src/utils/visualization.py | 38 +++++++++--------- tests/integration/test_api.py | 2 +- tests/unit/test_analysis.py | 16 +++++--- tests/unit/test_comparison.py | 27 ++++++------- 14 files changed, 270 insertions(+), 162 deletions(-) diff --git a/Ammeters/base_ammeter.py b/Ammeters/base_ammeter.py index 1159663..c5e9afa 100644 --- a/Ammeters/base_ammeter.py +++ b/Ammeters/base_ammeter.py @@ -6,8 +6,9 @@ NotImplementedErrorMsg = "Subclasses must implement this property." class AmmeterEmulatorBase(ABC): - def __init__(self, port: int): + def __init__(self, port: int, chaos_mode: bool = False): self.port = port + self.chaos_mode = chaos_mode random.seed(time.time()) # Seed the random number generator for each instance def start_server(self): @@ -27,9 +28,24 @@ def start_server(self): print(f"Connected by {addr}") data = conn.recv(1024) if data == self.get_current_command: - # Call the specific measure_current() method defined in subclasses - current = self.measure_current() - conn.sendall(str(current).encode('utf-8')) + if self.chaos_mode and random.random() < 0.10: + fault = random.choice(['sleep', 'garbage', 'close']) + if fault == 'sleep': + time.sleep(5) # Delay longer than typical timeout + current = self.measure_current() + try: + conn.sendall(str(current).encode('utf-8')) + except Exception: + pass + elif fault == 'garbage': + conn.sendall(b'ERR_NO_DATA') + elif fault == 'close': + # Abruptly close without sending + pass + else: + # Call the specific measure_current() method defined in subclasses + current = self.measure_current() + conn.sendall(str(current).encode('utf-8')) @property @abstractmethod diff --git a/README.md b/README.md index 76aa0f0..be4c32c 100644 --- a/README.md +++ b/README.md @@ -4,24 +4,31 @@ This project provides emulators for different types of ammeters: Greenlee, ENTES ## Project Structure +- `main.py`: Main script to start the ammeter emulators in the background. - `Ammeters/` - - `main.py`: Main script to start the ammeter emulators and request current measurements. - `Circutor_Ammeter.py`: Emulator for the CIRCUTOR ammeter. - `Entes_Ammeter.py`: Emulator for the ENTES ammeter. - `Greenlee_Ammeter.py`: Emulator for the Greenlee ammeter. - `base_ammeter.py`: Base class for all ammeter emulators. - `client.py`: Client to request current measurements from the ammeter emulators. - `config/` - - `config.yaml`: Configuration file for the ammeter emulators. + - `config.yaml`: Configuration file for the test framework and emulators. - `examples/` - - `run_test.py`: super lyze example for run test **don't use it**. + - `run_framework.py`: Production-ready CLI script to run automated tests. + - `compare_runs.py`: CLI script to compare two historical JSON archives. - `src/` - `testing/` - - `AmmeterTester.py`: Class to test the ammeter emulators. + - `test_framework.py`: Contains `AmmeterTestFramework`, the unified testing API and sampling engine. - `utils/` - - `config.py`: Configuration settings. - - `logger.py`: Logging setup. + - `config.py`: Configuration loader. + - `logger.py`: Logging setup and file handling. - `Utils.py`: Utility functions, including `generate_random_float`. + - `analysis.py`: Statistical calculation module. + - `visualization.py`: Matplotlib plotting module. + - `comparison.py`: Historical run comparison utility. +- `tests/` + - `integration/`: End-to-end pytest verification (e.g., `test_api.py`). + - `unit/`: Isolated pytest unit tests (e.g., `test_analysis.py`, `test_comparison.py`). ## Usage @@ -46,6 +53,45 @@ python3 main.py --- +## Sample Test Results + +When you execute a test run using the framework, it generates a comprehensive JSON report containing the raw data arrays, test metadata, and calculated statistics. Here is an example of a generated result file (`results/greenlee_20260506_145000_a1b2c3d4.json`): + +```json +{ + "ammeter_type": "greenlee", + "measurements": [ + 0.5010, + 5.8526, + 0.0742, + 0.0197, + 1.2961, + 5.7601, + 0.0731, + 0.0555, + 1.1312, + 0.0484 + ], + "count": 10, + "expected_count": 10, + "duration_seconds": 4.5129, + "sampling_frequency_hz": 2.0, + "statistics": { + "mean": 1.4812, + "median": 0.2876, + "min": 0.0197, + "max": 5.8526, + "stdev": 2.3266 + }, + "test_id": "a1b2c3d4-xxxx-xxxx-xxxx-xxxxxxxxxxxx", + "timestamp": "20260506_145000", + "plot_path": "results/greenlee_20260506_145000_plot.png", + "archive_path": "results/greenlee_20260506_145000_a1b2c3d4.json" +} +``` + +--- + ## Design Decisions & Bug Fixes ### 1. Emulator Communication Fix (PR #10) diff --git a/config/config.yaml b/config/config.yaml index b4c95d6..b3855cc 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -1,4 +1,5 @@ testing: + error_simulation: false sampling: measurements_count: 10 total_duration_seconds: 5 diff --git a/examples/compare_runs.py b/examples/compare_runs.py index d85584f..f14a895 100644 --- a/examples/compare_runs.py +++ b/examples/compare_runs.py @@ -1,9 +1,8 @@ -import os import sys -import argparse +from pathlib import Path -# Ensure the root directory is in the python path -sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) +# Ensure the root directory is in the path for imports +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from src.utils.comparison import compare_historical_runs from src.utils.config import load_config @@ -11,19 +10,20 @@ def main(): print("=== Ammeter Test Framework - Historical Comparison ===\n") - # 1. Load config to find the output directory dynamically - project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) - config_path = os.path.join(project_root, 'config', 'config.yaml') - config = load_config(config_path) - output_dir = config.get('result_management', {}).get('output_dir', 'results') + # Load config to find the output directory dynamically + project_root = Path(__file__).resolve().parent.parent + config_path = project_root / 'config' / 'config.yaml' + config = load_config(str(config_path)) + output_dir = Path(config.get('result_management', {}).get('output_dir', 'results')) - # 2. Setup CLI argument parser + # Setup CLI argument parser + import argparse parser = argparse.ArgumentParser(description="Compare two historical ammeter test runs.") parser.add_argument('--file1', type=str, help='Filename of the first test run') parser.add_argument('--file2', type=str, help='Filename of the second test run') args = parser.parse_args() - # 3. Interactive Fallback Logic + # Interactive Fallback Logic file1 = args.file1 file2 = args.file2 @@ -36,15 +36,41 @@ def main(): print("\nError: Both filenames must be provided. Exiting.") sys.exit(1) - # 4. Safely construct full paths - # If the user typed the full path, use it. Otherwise, prepend the output_dir. - path1 = file1 if os.path.isabs(file1) else os.path.join(output_dir, file1) - path2 = file2 if os.path.isabs(file2) else os.path.join(output_dir, file2) + # Safely construct full paths + path1 = Path(file1) if Path(file1).is_absolute() else output_dir / file1 + path2 = Path(file2) if Path(file2).is_absolute() else output_dir / file2 print(f"\nComparing [{file1}] vs [{file2}]...\n") - # 5. Execute Comparison - compare_historical_runs(path1, path2) + # Execute Comparison + try: + comparison_data = compare_historical_runs(str(path1), str(path2)) + + # Simple table formatting + run1 = comparison_data['run1'] + run2 = comparison_data['run2'] + + m1_mean = f"{run1.get('mean'):.4f}" if run1.get('mean') is not None else "N/A" + m2_mean = f"{run2.get('mean'):.4f}" if run2.get('mean') is not None else "N/A" + m1_max = f"{run1.get('max'):.4f}" if run1.get('max') is not None else "N/A" + m2_max = f"{run2.get('max'):.4f}" if run2.get('max') is not None else "N/A" + dur1 = f"{run1.get('duration', 0):.4f}" + dur2 = f"{run2.get('duration', 0):.4f}" + + print("\n" + "="*60) + print(f"{'Metric':<20} | {'Run 1':<15} | {'Run 2':<15}") + print("-" * 60) + print(f"{'Ammeter Type':<20} | {run1.get('ammeter') or 'N/A':<15} | {run2.get('ammeter') or 'N/A':<15}") + print(f"{'Count':<20} | {run1.get('count') or 0:<15} | {run2.get('count') or 0:<15}") + print(f"{'Duration (s)':<20} | {dur1:<15} | {dur2:<15}") + print(f"{'Mean (A)':<20} | {m1_mean:<15} | {m2_mean:<15}") + print(f"{'Max (A)':<20} | {m1_max:<15} | {m2_max:<15}") + print("="*60 + "\n") + + except FileNotFoundError as e: + print(f"Error: {e}") + except Exception as e: + print(f"Comparison failed: {e}") if __name__ == "__main__": main() diff --git a/examples/run_framework.py b/examples/run_framework.py index 3f3656a..e874241 100644 --- a/examples/run_framework.py +++ b/examples/run_framework.py @@ -1,10 +1,10 @@ import json import argparse -import os import sys +from pathlib import Path # Ensure the root directory is in the path for imports -sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from src.testing.test_framework import AmmeterTestFramework @@ -34,4 +34,4 @@ def main(): sys.exit(1) if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/main.py b/main.py index 9328507..8653142 100644 --- a/main.py +++ b/main.py @@ -5,32 +5,36 @@ from Ammeters.Entes_Ammeter import EntesAmmeter from Ammeters.Greenlee_Ammeter import GreenleeAmmeter from Ammeters.client import request_current_from_ammeter +from src.utils.config import load_config -def run_greenlee_emulator(): - greenlee = GreenleeAmmeter(5000) +def run_greenlee_emulator(chaos_mode: bool = False): + greenlee = GreenleeAmmeter(5000, chaos_mode=chaos_mode) greenlee.start_server() -def run_entes_emulator(): - entes = EntesAmmeter(5001) +def run_entes_emulator(chaos_mode: bool = False): + entes = EntesAmmeter(5001, chaos_mode=chaos_mode) entes.start_server() -def run_circutor_emulator(): - circutor = CircutorAmmeter(5002) +def run_circutor_emulator(chaos_mode: bool = False): + circutor = CircutorAmmeter(5002, chaos_mode=chaos_mode) circutor.start_server() -def start_emulators(): +def start_emulators(chaos_mode: bool = False): # Start each ammeter in a separate thread - threading.Thread(target=run_greenlee_emulator, daemon=True).start() - threading.Thread(target=run_entes_emulator, daemon=True).start() - threading.Thread(target=run_circutor_emulator, daemon=True).start() + threading.Thread(target=run_greenlee_emulator, args=(chaos_mode,), daemon=True).start() + threading.Thread(target=run_entes_emulator, args=(chaos_mode,), daemon=True).start() + threading.Thread(target=run_circutor_emulator, args=(chaos_mode,), daemon=True).start() # Wait for the servers to start, if you have problem restarting the servers between runs try increasing sleep time. time.sleep(5) if __name__ == "__main__": - print("Starting ammeter emulators...") - start_emulators() + config = load_config("config/config.yaml") + chaos_mode = config.get("testing", {}).get("error_simulation", False) + + print(f"Starting ammeter emulators... (Chaos Mode: {chaos_mode})") + start_emulators(chaos_mode) # Request an initial reading just to verify connection request_current_from_ammeter(5000, b'MEASURE_GREENLEE -get_measurement') diff --git a/src/testing/test_framework.py b/src/testing/test_framework.py index a1b3808..13a7938 100644 --- a/src/testing/test_framework.py +++ b/src/testing/test_framework.py @@ -2,8 +2,8 @@ import time import json -import os import uuid +from pathlib import Path from datetime import datetime from typing import Optional import typing @@ -11,7 +11,9 @@ from ..utils.analysis import calculate_statistics from ..utils.visualization import generate_simple_plot from Ammeters.client import request_current_from_ammeter +from ..utils.logger import TestLogger +logger = TestLogger("AmmeterTestFramework").logger class AmmeterTestFramework: def __init__(self, config_path: str = "config/config.yaml"): @@ -49,26 +51,24 @@ def run_test(self, ammeter_type: str) -> dict: """ sampling_cfg = self.config.get('testing', {}).get('sampling', {}) - # Extract raw configuration values - count = sampling_cfg.get('measurements_count') - duration = sampling_cfg.get('total_duration_seconds') - frequency = sampling_cfg.get('sampling_frequency_hz') + # Helper to convert 'NULL' string or None to real Python None + def normalize(val): + return None if val is None or val == 'NULL' else val + + count = normalize(sampling_cfg.get('measurements_count')) + duration = normalize(sampling_cfg.get('total_duration_seconds')) + frequency = normalize(sampling_cfg.get('sampling_frequency_hz')) # Calculate the delay between samples based on the target frequency - freq_val = float(frequency) if frequency and frequency != 'NULL' else 1.0 + freq_val = float(frequency) if frequency is not None else 1.0 delay = 1.0 / freq_val - # Determine which constraints have been explicitly configured - limit_by_count = count is not None and count != 'NULL' - limit_by_duration = duration is not None and duration != 'NULL' - - # Enforce that the test is bounded by at least one constraint to prevent infinite loops - if not limit_by_count and not limit_by_duration: - raise ValueError("Both measurements_count and total_duration_seconds are missing or NULL. At least one must be provided.") + if count is None and duration is None: + raise ValueError("Both measurements_count and total_duration_seconds are missing. At least one must be provided.") # Convert limits to infinity if they are not configured, so the while loop ignores them - max_count = int(count) if limit_by_count else float('inf') - max_duration = float(duration) if limit_by_duration else float('inf') + max_count = int(count) if count is not None else float('inf') + max_duration = float(duration) if duration is not None else float('inf') measurements: typing.List[float] = [] start_time = time.time() @@ -78,15 +78,16 @@ def run_test(self, ammeter_type: str) -> dict: val = self.get_single_reading(ammeter_type) if val is not None: measurements.append(val) + logger.debug(f"Captured reading: {val}") # Prevent an unnecessary trailing sleep delay if we've just hit the exact count limit if len(measurements) >= max_count: break - # Wait for the next sampling cycle to maintain the requested frequency time.sleep(delay) actual_duration = time.time() - start_time + logger.info(f"Test run completed for {ammeter_type}. Collected {len(measurements)} samples.") result = { "ammeter_type": ammeter_type, @@ -101,45 +102,39 @@ def run_test(self, ammeter_type: str) -> dict: def _process_results(self, result: dict) -> dict: """Helper method to handle statistical calculations, visualization, and JSON archiving.""" - # Generate unique test ID and timestamp - test_id = str(uuid.uuid4()) - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - - result['test_id'] = test_id - result['timestamp'] = timestamp - analysis_cfg = self.config.get('analysis', {}) measurements: typing.List[float] = result.get('measurements', []) - # Calculate statistics if enabled (will explicitly fail if measurements is empty) if analysis_cfg.get('statistical_metrics'): result['statistics'] = calculate_statistics(measurements) - # Output directory resolution - output_dir = self.config.get('result_management', {}).get('output_dir', 'results') if self.config.get('result_management') else 'results' + output_dir = Path(self.config.get('result_management', {}).get('output_dir', 'results')) - # Generate visualization if enabled (will explicitly fail if measurements is empty) vis_cfg = analysis_cfg.get('visualization', {}) if vis_cfg.get('enabled'): stats = result.get('statistics') - plot_path = generate_simple_plot( ammeter_type=result['ammeter_type'], measurements=measurements, - output_dir=output_dir, + output_dir=str(output_dir), stats=stats ) if plot_path: result['plot_path'] = plot_path - # Result Management: Archive the test run to a structured JSON file - os.makedirs(output_dir, exist_ok=True) + output_dir.mkdir(parents=True, exist_ok=True) + test_id = str(uuid.uuid4()) + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + result['test_id'] = test_id + result['timestamp'] = timestamp + json_filename = f"{result['ammeter_type']}_{timestamp}_{test_id[:8]}.json" - json_filepath = os.path.join(output_dir, json_filename) + json_filepath = output_dir / json_filename with open(json_filepath, 'w', encoding='utf-8') as f: json.dump(result, f, indent=2, ensure_ascii=False) - result['archive_path'] = json_filepath + result['archive_path'] = str(json_filepath) + logger.info(f"Results archived to {json_filepath}") return result diff --git a/src/utils/analysis.py b/src/utils/analysis.py index a82b392..f1796f5 100644 --- a/src/utils/analysis.py +++ b/src/utils/analysis.py @@ -1,15 +1,23 @@ import statistics -from typing import Dict, List +from typing import Dict, List, Any -def calculate_statistics(measurements: List[float]) -> Dict[str, float]: +def calculate_statistics(measurements: List[float], cv_threshold: float = 5.0) -> Dict[str, Any]: """Calculates statistical metrics for a given list of measurements.""" if not measurements: raise ValueError("Cannot calculate statistics on an empty list. Please increase the measurement count or duration to collect data.") + mean_val = statistics.mean(measurements) + stdev_val = statistics.stdev(measurements) if len(measurements) > 1 else 0.0 + + cv_percentage = (stdev_val / mean_val * 100) if mean_val != 0 else 0.0 + is_consistent = cv_percentage <= cv_threshold + return { - "mean": statistics.mean(measurements), + "mean": mean_val, "median": statistics.median(measurements), "min": min(measurements), "max": max(measurements), - "stdev": statistics.stdev(measurements) if len(measurements) > 1 else 0.0 + "stdev": stdev_val, + "cv_percentage": cv_percentage, + "is_consistent": is_consistent } diff --git a/src/utils/comparison.py b/src/utils/comparison.py index 892c063..6de4cc4 100644 --- a/src/utils/comparison.py +++ b/src/utils/comparison.py @@ -1,50 +1,47 @@ import json -import os +from pathlib import Path +from typing import Dict, Any -def compare_historical_runs(file_path_1: str, file_path_2: str) -> None: +def compare_historical_runs(file_path_1: str, file_path_2: str) -> Dict[str, Any]: """Compares two historical test runs side-by-side using their file paths. Args: file_path_1: Path to the first JSON result file. file_path_2: Path to the second JSON result file. + + Returns: + A dictionary containing the parsed run data for comparison. + + Raises: + FileNotFoundError: If either file does not exist. + json.JSONDecodeError: If either file contains invalid JSON. """ - if not os.path.exists(file_path_1): - print(f"Error: Could not find file -> {file_path_1}") - return - if not os.path.exists(file_path_2): - print(f"Error: Could not find file -> {file_path_2}") - return - - try: - with open(file_path_1, 'r', encoding='utf-8') as f1: - run1_data = json.load(f1) - except json.JSONDecodeError: - print(f"Error: Invalid JSON format in {file_path_1}") - return - - try: - with open(file_path_2, 'r', encoding='utf-8') as f2: - run2_data = json.load(f2) - except json.JSONDecodeError: - print(f"Error: Invalid JSON format in {file_path_2}") - return + path1 = Path(file_path_1) + path2 = Path(file_path_2) + + if not path1.exists() or not path2.exists(): + raise FileNotFoundError("One or both historical run files not found.") + + with path1.open('r', encoding='utf-8') as f1, path2.open('r', encoding='utf-8') as f2: + run1_data = json.load(f1) + run2_data = json.load(f2) stats1 = run1_data.get('statistics', {}) stats2 = run2_data.get('statistics', {}) - m1_mean = f"{stats1.get('mean'):.4f}" if isinstance(stats1.get('mean'), float) else "N/A" - m2_mean = f"{stats2.get('mean'):.4f}" if isinstance(stats2.get('mean'), float) else "N/A" - m1_max = f"{stats1.get('max'):.4f}" if isinstance(stats1.get('max'), float) else "N/A" - m2_max = f"{stats2.get('max'):.4f}" if isinstance(stats2.get('max'), float) else "N/A" - dur1 = f"{run1_data.get('duration_seconds', 0):.4f}" - dur2 = f"{run2_data.get('duration_seconds', 0):.4f}" - - print("\n" + "="*60) - print(f"{'Metric':<20} | {'Run 1':<15} | {'Run 2':<15}") - print("-" * 60) - print(f"{'Ammeter Type':<20} | {run1_data.get('ammeter_type', 'N/A'):<15} | {run2_data.get('ammeter_type', 'N/A'):<15}") - print(f"{'Count':<20} | {run1_data.get('count', 0):<15} | {run2_data.get('count', 0):<15}") - print(f"{'Duration (s)':<20} | {dur1:<15} | {dur2:<15}") - print(f"{'Mean (A)':<20} | {m1_mean:<15} | {m2_mean:<15}") - print(f"{'Max (A)':<20} | {m1_max:<15} | {m2_max:<15}") - print("="*60 + "\n") \ No newline at end of file + return { + "run1": { + "ammeter": run1_data.get('ammeter_type'), + "count": run1_data.get('count'), + "duration": run1_data.get('duration_seconds'), + "mean": stats1.get('mean'), + "max": stats1.get('max'), + }, + "run2": { + "ammeter": run2_data.get('ammeter_type'), + "count": run2_data.get('count'), + "duration": run2_data.get('duration_seconds'), + "mean": stats2.get('mean'), + "max": stats2.get('max'), + } + } diff --git a/src/utils/logger.py b/src/utils/logger.py index 3f62642..21aad36 100644 --- a/src/utils/logger.py +++ b/src/utils/logger.py @@ -1,5 +1,5 @@ import logging -import os +from pathlib import Path from datetime import datetime class TestLogger: @@ -12,19 +12,31 @@ def _setup_logger(self) -> logging.Logger: הגדרת הלוגר עם פורמט מותאם וכתיבה לקובץ """ # יצירת תיקיית הלוגים - log_dir = "results/logs" - os.makedirs(log_dir, exist_ok=True) + log_dir = Path("results/logs") + log_dir.mkdir(parents=True, exist_ok=True) # הגדרת שם הקובץ עם תאריך ומזהה הבדיקה timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - log_file = f"{log_dir}/{timestamp}_{self._test_name}.log" + log_file = log_dir / f"{timestamp}_{self._test_name}.log" # Using log_file to solve the linter warning about unused variable # Didn't want to change whatever I want and touch the existing code if not necessary print(f"Logging to: {log_file}") # הגדרת הלוגר logger = logging.getLogger(f"test_{self._test_name}") + logger.setLevel(logging.DEBUG) + if not logger.handlers: + # Create a file handler + fh = logging.FileHandler(log_file) + fh.setLevel(logging.DEBUG) + + # Create a formatter and add it to the handler + formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') + fh.setFormatter(formatter) + + # Add the handler to the logger + logger.addHandler(fh) return logger @@ -38,4 +50,4 @@ def debug(self, message: str): self.logger.debug(message) def warning(self, message: str): - self.logger.warning(message) \ No newline at end of file + self.logger.warning(message) diff --git a/src/utils/visualization.py b/src/utils/visualization.py index 2f3c10b..6cab734 100644 --- a/src/utils/visualization.py +++ b/src/utils/visualization.py @@ -1,47 +1,47 @@ -import os -import matplotlib.pyplot as plt # type: ignore +import matplotlib.pyplot as plt +from pathlib import Path from typing import List, Optional, Dict from datetime import datetime def generate_simple_plot(ammeter_type: str, measurements: List[float], output_dir: str = "results", stats: Optional[Dict[str, float]] = None) -> str: """ Generates a simple, clean, and user-friendly line plot of the current measurements over time. - Acts like a dashboard panel by including optional statistical reference lines. + Uses Matplotlib's Object-Oriented API for thread-safe plotting. """ if not measurements: return "" - os.makedirs(output_dir, exist_ok=True) + output_path = Path(output_dir) + output_path.mkdir(parents=True, exist_ok=True) - plt.figure(figsize=(10, 5)) + fig, ax = plt.subplots(figsize=(10, 5)) - # Simple line plot with markers for clean visualization - plt.plot(measurements, marker='o', linestyle='-', color='#007acc', linewidth=2, markersize=6, label="Current (A)") + # Simple line plot with markers + ax.plot(measurements, marker='o', linestyle='-', color='#007acc', linewidth=2, markersize=6, label="Current (A)") # Dashboard-Style Visualization: Add reference lines if stats are provided if stats: if 'mean' in stats: - plt.axhline(y=stats['mean'], color='green', linestyle='--', linewidth=2, label=f"Mean: {stats['mean']:.4f} A") + ax.axhline(y=stats['mean'], color='green', linestyle='--', linewidth=2, label=f"Mean: {stats['mean']:.4f} A") if 'max' in stats: - plt.axhline(y=stats['max'], color='red', linestyle=':', linewidth=2, label=f"Max: {stats['max']:.4f} A") + ax.axhline(y=stats['max'], color='red', linestyle=':', linewidth=2, label=f"Max: {stats['max']:.4f} A") - # Add legend to explain the reference lines - plt.legend(loc='best', fontsize=10) + ax.legend(loc='best', fontsize=10) - plt.title(f"Current Measurements over Time: {ammeter_type.upper()}", fontsize=14, pad=15) - plt.xlabel("Sample Index", fontsize=12) - plt.ylabel("Current (A)", fontsize=12) + ax.set_title(f"Current Measurements over Time: {ammeter_type.upper()}", fontsize=14, pad=15) + ax.set_xlabel("Sample Index", fontsize=12) + ax.set_ylabel("Current (A)", fontsize=12) # User-friendly grid for better readability - plt.grid(True, linestyle='--', alpha=0.6) + ax.grid(True, linestyle='--', alpha=0.6) plt.tight_layout() timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"{ammeter_type}_{timestamp}_plot.png" - filepath = os.path.join(output_dir, filename) + filepath = output_path / filename - plt.savefig(filepath) - plt.close() + fig.savefig(filepath) + plt.close(fig) - return filepath + return str(filepath) diff --git a/tests/integration/test_api.py b/tests/integration/test_api.py index 3837f4b..fe898cf 100644 --- a/tests/integration/test_api.py +++ b/tests/integration/test_api.py @@ -111,5 +111,5 @@ def test_run_test_both_null_raises_value_error(framework, original_config): 'sampling_frequency_hz': 10 }) - with pytest.raises(ValueError, match="Both measurements_count and total_duration_seconds are missing or NULL"): + with pytest.raises(ValueError, match="Both measurements_count and total_duration_seconds are missing. At least one must be provided."): framework.run_test('greenlee') diff --git a/tests/unit/test_analysis.py b/tests/unit/test_analysis.py index 63ea236..211cd08 100644 --- a/tests/unit/test_analysis.py +++ b/tests/unit/test_analysis.py @@ -2,14 +2,20 @@ from src.utils.analysis import calculate_statistics @pytest.mark.parametrize("measurements, expected", [ - ([1.0, 2.0, 3.0, 4.0, 5.0], {"mean": 3.0, "median": 3.0, "min": 1.0, "max": 5.0, "stdev": 1.5811388}), - ([10.5], {"mean": 10.5, "median": 10.5, "min": 10.5, "max": 10.5, "stdev": 0.0}), - ([-1.0, -2.0, -3.0], {"mean": -2.0, "median": -2.0, "min": -3.0, "max": -1.0, "stdev": 1.0}), - ([1e10, 1e-10], {"mean": 5e9, "median": 5e9, "min": 1e-10, "max": 1e10, "stdev": 7071067811.865475}), + ([1.0, 2.0, 3.0, 4.0, 5.0], {"mean": 3.0, "median": 3.0, "min": 1.0, "max": 5.0, "stdev": 1.5811388, "cv_percentage": 52.7046276, "is_consistent": False}), + ([10.5], {"mean": 10.5, "median": 10.5, "min": 10.5, "max": 10.5, "stdev": 0.0, "cv_percentage": 0.0, "is_consistent": True}), + ([-1.0, -2.0, -3.0], {"mean": -2.0, "median": -2.0, "min": -3.0, "max": -1.0, "stdev": 1.0, "cv_percentage": -50.0, "is_consistent": True}), + ([1e10, 1e-10], {"mean": 5e9, "median": 5e9, "min": 1e-10, "max": 1e10, "stdev": 7071067811.865475, "cv_percentage": 141.421356, "is_consistent": False}), ]) def test_calculate_statistics(measurements, expected): stats = calculate_statistics(measurements) - assert stats == pytest.approx(expected, rel=1e-5) + + # Check boolean explicitly, then remove before approx to avoid strict type issues + assert stats["is_consistent"] == expected["is_consistent"] + del stats["is_consistent"] + + expected_floats = {k: v for k, v in expected.items() if k != "is_consistent"} + assert stats == pytest.approx(expected_floats, rel=1e-5) def test_calculate_statistics_empty(): with pytest.raises(ValueError, match="Cannot calculate statistics"): diff --git a/tests/unit/test_comparison.py b/tests/unit/test_comparison.py index a82f8af..57273a3 100644 --- a/tests/unit/test_comparison.py +++ b/tests/unit/test_comparison.py @@ -23,29 +23,26 @@ def _create(test_id, ammeter_type="greenlee", stats=None): return str(file_path) return _create -def test_compare_historical_runs_happy_path(create_mock_run, capsys): +def test_compare_historical_runs_happy_path(create_mock_run): stats = {"mean": 0.5, "max": 1.0} path1 = create_mock_run("id1", "greenlee", stats) path2 = create_mock_run("id2", "greenlee", stats) - compare_historical_runs(path1, path2) + data = compare_historical_runs(path1, path2) - captured = capsys.readouterr() - assert "Metric" in captured.out - assert "Run 1" in captured.out - assert "Run 2" in captured.out - assert "Mean (A)" in captured.out + assert data["run1"]["ammeter"] == "greenlee" + assert data["run1"]["mean"] == 0.5 + assert data["run2"]["mean"] == 0.5 -def test_compare_historical_runs_file_not_found(capsys): - compare_historical_runs("non_existent_1.json", "non_existent_2.json") - captured = capsys.readouterr() - assert "Error: Could not find file" in captured.out +def test_compare_historical_runs_file_not_found(): + with pytest.raises(FileNotFoundError): + compare_historical_runs("non_existent_1.json", "non_existent_2.json") -def test_compare_historical_runs_missing_stats(create_mock_run, capsys): +def test_compare_historical_runs_missing_stats(create_mock_run): path1 = create_mock_run("id1", "greenlee", stats=None) path2 = create_mock_run("id2", "greenlee", stats=None) - compare_historical_runs(path1, path2) + data = compare_historical_runs(path1, path2) - captured = capsys.readouterr() - assert "N/A" in captured.out + assert data["run1"]["mean"] is None + assert data["run2"]["mean"] is None From 2d6aec7dfc35287f49dc1dde1fc3b0cae4ec816f Mon Sep 17 00:00:00 2001 From: netanelC Date: Wed, 6 May 2026 23:37:52 +0300 Subject: [PATCH 2/2] lint: add type ignore Co-authored-by: Copilot --- src/utils/visualization.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/utils/visualization.py b/src/utils/visualization.py index 6cab734..4a2fc2c 100644 --- a/src/utils/visualization.py +++ b/src/utils/visualization.py @@ -1,4 +1,4 @@ -import matplotlib.pyplot as plt +import matplotlib.pyplot as plt # type: ignore from pathlib import Path from typing import List, Optional, Dict from datetime import datetime