From d5d53fd42a891192c87ecad41c76877c52f45244 Mon Sep 17 00:00:00 2001 From: netanelC Date: Wed, 6 May 2026 20:32:03 +0300 Subject: [PATCH 1/2] feat: result management Co-authored-by: Copilot --- README.md | 11 +++++++- config/config.yaml | 1 + examples/compare_runs.py | 50 +++++++++++++++++++++++++++++++++ examples/run_framework.py | 37 +++++++++++++++++++++++++ src/testing/test_framework.py | 32 +++++++++++++++++---- src/utils/comparison.py | 50 +++++++++++++++++++++++++++++++++ tests/integration/test_api.py | 17 ++++++++++++ tests/unit/test_comparison.py | 52 +++++++++++++++++++++++++++++++++++ 8 files changed, 244 insertions(+), 6 deletions(-) create mode 100644 examples/compare_runs.py create mode 100644 examples/run_framework.py create mode 100644 src/utils/comparison.py create mode 100644 tests/unit/test_comparison.py diff --git a/README.md b/README.md index 37338f1..b9b97db 100644 --- a/README.md +++ b/README.md @@ -100,4 +100,13 @@ python3 main.py - Created a robust statistical module in `src/utils/analysis.py` leveraging Python's built-in `statistics` library to compute Mean, Median, Standard Deviation, Min, and Max. - Integrated a clean, dashboard-style visualization module in `src/utils/visualization.py` utilizing `matplotlib`. It automatically generates line plots with user-friendly grids and includes optional horizontal reference lines for `Mean` and `Max` values. - **Decoupled Architecture:** Extracted the statistical calculation and visualization I/O logic out of the monolithic `run_test()` method into a private `_process_results()` helper method. -- **Test Artifact Cleanup:** Updated the integration test suite to utilize pytest's built-in `tmp_path` fixture. The framework dynamically overrides its `output_dir` during testing so plots are written to ephemeral directories and automatically cleaned up, keeping the workspace pristine. \ No newline at end of file +- **Test Artifact Cleanup:** Updated the integration test suite to utilize pytest's built-in `tmp_path` fixture. The framework dynamically overrides its `output_dir` during testing so plots are written to ephemeral directories and automatically cleaned up, keeping the workspace pristine. + +### 6. Result Management and Archiving System (Issue #13) +**The Problem:** The exam requires a robust result archiving system to store test runs for historical review and comparison. +**The Design:** +- Enhanced `_process_results()` to generate a unique `test_id` (UUID) and `timestamp` for every automated test run. +- Archiving mechanism dumps the entire test run metadata—including configuration parameters, raw measurements arrays, and calculated statistical metrics—to a structured JSON file. +- The JSON output is saved into the configurable `results/` directory using an identifiable naming convention (`{ammeter_type}_{timestamp}_{uuid}.json`). +- **Historical Comparison (Decoupled):** Adhering to the Single Responsibility Principle, the comparison logic was extracted into a standalone utility module (`src/utils/comparison.py`). It loads two archived JSON files by their paths and outputs a side-by-side terminal table comparing their counts, durations, and statistical metrics (Mean, Max) for clear, actionable observability. An executable example script is provided at `examples/compare_runs.py`. +- Updated the integration test suite to assert that the `archive_path` exists, the JSON structure maintains integrity, and the decoupled comparison table successfully generates. \ No newline at end of file diff --git a/config/config.yaml b/config/config.yaml index c187c2a..b4c95d6 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -23,3 +23,4 @@ analysis: - line result_management: + output_dir: "results" diff --git a/examples/compare_runs.py b/examples/compare_runs.py new file mode 100644 index 0000000..0ddfffe --- /dev/null +++ b/examples/compare_runs.py @@ -0,0 +1,50 @@ +import os +import sys +import argparse + +# Ensure the root directory is in the python path +project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '../')) +sys.path.insert(0, project_root) + +from src.utils.comparison import compare_historical_runs +from src.utils.config import load_config + +def main(): + print("=== Ammeter Test Framework - Historical Comparison ===\n") + + # 1. Load config to find the output directory dynamically + 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') + + # 2. Setup CLI argument parser + 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 + file1 = args.file1 + file2 = args.file2 + + if not file1: + file1 = input(f"Enter the filename for Run 1 (located in {output_dir}/): ").strip() + if not file2: + file2 = input(f"Enter the filename for Run 2 (located in {output_dir}/): ").strip() + + if not file1 or not file2: + 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) + + print(f"\nComparing [{file1}] vs [{file2}]...\n") + + # 5. Execute Comparison + compare_historical_runs(path1, path2) + +if __name__ == "__main__": + main() diff --git a/examples/run_framework.py b/examples/run_framework.py new file mode 100644 index 0000000..3f3656a --- /dev/null +++ b/examples/run_framework.py @@ -0,0 +1,37 @@ +import json +import argparse +import os +import sys + +# Ensure the root directory is in the path for imports +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) + +from src.testing.test_framework import AmmeterTestFramework + +def main(): + parser = argparse.ArgumentParser(description="Production-ready test runner for the Ammeter Framework") + parser.add_argument('--ammeter', type=str, default='greenlee', choices=['greenlee', 'entes', 'circutor'], + help="Which ammeter to test (default: greenlee)") + args = parser.parse_args() + + print(f"--- Running test against {args.ammeter.upper()} ---") + try: + framework = AmmeterTestFramework() + result = framework.run_test(args.ammeter) + + print("\n" + "="*50) + print("TEST RESULTS:") + print("="*50) + print(json.dumps(result, indent=2, default=str)) + + if 'plot_path' in result: + print(f"\n📊 Visualization saved to: {result['plot_path']}") + if 'archive_path' in result: + print(f"💾 Results archived at: {result['archive_path']}") + + except Exception as e: + print(f"\n❌ Framework execution failed: {e}") + sys.exit(1) + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/src/testing/test_framework.py b/src/testing/test_framework.py index cafa671..a1b3808 100644 --- a/src/testing/test_framework.py +++ b/src/testing/test_framework.py @@ -1,6 +1,10 @@ import time +import json +import os +import uuid +from datetime import datetime from typing import Optional import typing from ..utils.config import load_config @@ -96,19 +100,27 @@ def run_test(self, ammeter_type: str) -> dict: return self._process_results(result) def _process_results(self, result: dict) -> dict: - """Helper method to handle statistical calculations and visualization I/O cleanly.""" + """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 and we have actual data + # Calculate statistics if enabled (will explicitly fail if measurements is empty) if analysis_cfg.get('statistical_metrics'): result['statistics'] = calculate_statistics(measurements) - # Generate visualization if enabled and we have actual data + # Output directory resolution + output_dir = self.config.get('result_management', {}).get('output_dir', 'results') if self.config.get('result_management') else 'results' + + # Generate visualization if enabled (will explicitly fail if measurements is empty) vis_cfg = analysis_cfg.get('visualization', {}) if vis_cfg.get('enabled'): - # Extract output directory safely, defaulting to 'results' - output_dir = self.config.get('result_management', {}).get('output_dir', 'results') if self.config.get('result_management') else 'results' stats = result.get('statistics') plot_path = generate_simple_plot( @@ -120,4 +132,14 @@ def _process_results(self, result: dict) -> dict: 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) + json_filename = f"{result['ammeter_type']}_{timestamp}_{test_id[:8]}.json" + json_filepath = os.path.join(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 + return result diff --git a/src/utils/comparison.py b/src/utils/comparison.py new file mode 100644 index 0000000..892c063 --- /dev/null +++ b/src/utils/comparison.py @@ -0,0 +1,50 @@ +import json +import os + +def compare_historical_runs(file_path_1: str, file_path_2: str) -> None: + """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. + """ + 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 + + 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 diff --git a/tests/integration/test_api.py b/tests/integration/test_api.py index 9b1c248..3837f4b 100644 --- a/tests/integration/test_api.py +++ b/tests/integration/test_api.py @@ -64,6 +64,23 @@ def test_run_test_count_before_duration(framework, original_config, tmp_path): assert 'plot_path' in res assert os.path.exists(res['plot_path']) + + # Verify archiving system metadata + assert 'test_id' in res + assert 'timestamp' in res + assert 'archive_path' in res + + archive_path = res['archive_path'] + assert os.path.exists(archive_path) + + # Verify the JSON file contains the correct test structure + import json + with open(archive_path, 'r', encoding='utf-8') as f: + archived_data = json.load(f) + + assert archived_data['test_id'] == res['test_id'] + assert archived_data['ammeter_type'] == 'greenlee' + assert len(archived_data['measurements']) == 2 def test_run_test_duration_before_count(framework, original_config, tmp_path): # Setup framework to hit duration limit before count limit diff --git a/tests/unit/test_comparison.py b/tests/unit/test_comparison.py new file mode 100644 index 0000000..e3b3613 --- /dev/null +++ b/tests/unit/test_comparison.py @@ -0,0 +1,52 @@ +import json +import pytest +import os +from src.utils.comparison import compare_historical_runs + +@pytest.fixture +def create_mock_run(tmp_path): + """Helper to create a mock JSON result file with the correct naming convention.""" + def _create(test_id, ammeter_type="greenlee", stats=None): + data = { + "test_id": test_id, + "ammeter_type": ammeter_type, + "count": 10, + "duration_seconds": 5.0, + } + if stats is not None: + data["statistics"] = stats + + # Mimic the real filename structure: {type}_{timestamp}_{id}.json + file_name = f"{ammeter_type}_20260506_120000_{test_id}.json" + file_path = tmp_path / file_name + with open(file_path, 'w', encoding='utf-8') as f: + json.dump(data, f) + return str(file_path) + return _create + +def test_compare_historical_runs_happy_path(create_mock_run, capsys): + 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) + + 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 + +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_missing_stats(create_mock_run, capsys): + path1 = create_mock_run("id1", "greenlee", stats=None) + path2 = create_mock_run("id2", "greenlee", stats=None) + + compare_historical_runs(path1, path2) + + captured = capsys.readouterr() + assert "N/A" in captured.out From 92ec53bce3f56192f7bda6903171648792bc0aa0 Mon Sep 17 00:00:00 2001 From: netanelC Date: Wed, 6 May 2026 20:41:34 +0300 Subject: [PATCH 2/2] lint: fix lints --- README.md | 33 ++++++++++++--------------------- examples/compare_runs.py | 4 ++-- main.py | 19 ++++++++++++++----- tests/unit/test_comparison.py | 1 - 4 files changed, 28 insertions(+), 29 deletions(-) diff --git a/README.md b/README.md index b9b97db..76aa0f0 100644 --- a/README.md +++ b/README.md @@ -25,30 +25,21 @@ This project provides emulators for different types of ammeters: Greenlee, ENTES ## Usage -# Ammeter Emulators - -## Greenlee Ammeter - -- **Port**: 5000 -- **Command**: `MEASURE_GREENLEE -get_measurement` -- **Measurement Logic**: Calculates current using voltage (1V - 10V) and (0.1Ω - 100Ω). -- **Measurement method** : Ohm's Law: I = V / R - -## ENTES Ammeter - -- **Port**: 5001 -- **Command**: `MEASURE_ENTES -get_data` -- **Measurement Logic**: Calculates current using magnetic field strength (0.01T - 0.1T) and calibration factor (500 - 2000). -- **Measurement method** : Hall Effect: I = B * K +The project includes production-ready example scripts for running tests and comparing historical results: -## CIRCUTOR Ammeter +- **Run a new test:** + ```sh + python3 examples/run_framework.py --ammeter greenlee + ``` + This will execute the framework using settings in `config.yaml`, print the results in JSON format, and save the visualization/archive in the `results/` folder. -- **Port**: 5002 -- **Command**: `MEASURE_CIRCUTOR -get_measurement` -- **Measurement Logic**: Calculates current using voltage values (0.1V - 1.0V) over a number of samples and a random time step (0.001s - 0.01s). -- **Measurement method** : Rogowski Coil Integration: I = ∫V dt +- **Compare historical runs:** + ```sh + python3 examples/compare_runs.py + ``` + This script will prompt you for the filenames of two archived JSON result files in the `results/` folder and output a side-by-side terminal comparison of their statistics. -To start the ammeter emulators and request current measurements, run the `main.py` script: +To start the ammeter emulators in the background: ```sh python3 main.py ``` diff --git a/examples/compare_runs.py b/examples/compare_runs.py index 0ddfffe..d85584f 100644 --- a/examples/compare_runs.py +++ b/examples/compare_runs.py @@ -3,8 +3,7 @@ import argparse # Ensure the root directory is in the python path -project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '../')) -sys.path.insert(0, project_root) +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from src.utils.comparison import compare_historical_runs from src.utils.config import load_config @@ -13,6 +12,7 @@ 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') diff --git a/main.py b/main.py index f4947d7..9328507 100644 --- a/main.py +++ b/main.py @@ -29,10 +29,19 @@ def start_emulators(): time.sleep(5) if __name__ == "__main__": + print("Starting ammeter emulators...") start_emulators() - request_current_from_ammeter(5000, b'MEASURE_GREENLEE -get_measurement') # Request from Greenlee Ammeter - request_current_from_ammeter(5001, b'MEASURE_ENTES -get_data') # Request from ENTES Ammeter - request_current_from_ammeter(5002, b'MEASURE_CIRCUTOR -get_measurement') # Request from CIRCUTOR Ammeter - - pass + # Request an initial reading just to verify connection + request_current_from_ammeter(5000, b'MEASURE_GREENLEE -get_measurement') + request_current_from_ammeter(5001, b'MEASURE_ENTES -get_data') + request_current_from_ammeter(5002, b'MEASURE_CIRCUTOR -get_measurement') + + print("\nEmulators are running in the background. Press Ctrl+C to stop.") + try: + # Keep the main thread alive indefinitely so the daemon threads (emulators) stay up + while True: + time.sleep(1) + except KeyboardInterrupt: + print("\nShutting down emulators.") + pass diff --git a/tests/unit/test_comparison.py b/tests/unit/test_comparison.py index e3b3613..a82f8af 100644 --- a/tests/unit/test_comparison.py +++ b/tests/unit/test_comparison.py @@ -1,6 +1,5 @@ import json import pytest -import os from src.utils.comparison import compare_historical_runs @pytest.fixture