diff --git a/Ammeters/base_ammeter.py b/Ammeters/base_ammeter.py index c5e9afa..730bbd9 100644 --- a/Ammeters/base_ammeter.py +++ b/Ammeters/base_ammeter.py @@ -2,6 +2,7 @@ import time import random from abc import ABC, abstractmethod +from src.utils.logger import TestLogger NotImplementedErrorMsg = "Subclasses must implement this property." @@ -9,6 +10,7 @@ class AmmeterEmulatorBase(ABC): def __init__(self, port: int, chaos_mode: bool = False): self.port = port self.chaos_mode = chaos_mode + self.logger = TestLogger(self.__class__.__name__).logger random.seed(time.time()) # Seed the random number generator for each instance def start_server(self): @@ -21,11 +23,11 @@ def start_server(self): s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind(('localhost', self.port)) s.listen() - print(f"{self.__class__.__name__} is running on port {self.port}") + self.logger.info(f"{self.__class__.__name__} is running on port {self.port}") while True: conn, addr = s.accept() with conn: - print(f"Connected by {addr}") + self.logger.info(f"Connected by {addr}") data = conn.recv(1024) if data == self.get_current_command: if self.chaos_mode and random.random() < 0.10: @@ -63,4 +65,3 @@ def measure_current(self) -> float: logic for current measurement. """ raise NotImplementedError(NotImplementedErrorMsg) - diff --git a/README.md b/README.md index be4c32c..4b76912 100644 --- a/README.md +++ b/README.md @@ -4,52 +4,96 @@ 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/` - - `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 test framework and emulators. -- `examples/` - - `run_framework.py`: Production-ready CLI script to run automated tests. - - `compare_runs.py`: CLI script to compare two historical JSON archives. -- `src/` - - `testing/` - - `test_framework.py`: Contains `AmmeterTestFramework`, the unified testing API and sampling engine. - - `utils/` - - `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 - -The project includes production-ready example scripts for running tests and comparing historical results: - -- **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. - -- **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 in the background: +```text +ammeter-test-framework/ +├── Ammeters/ +│ ├── Circutor_Ammeter.py +│ ├── Entes_Ammeter.py +│ ├── Greenlee_Ammeter.py +│ ├── base_ammeter.py +│ └── client.py +├── config/ +│ └── config.yaml +├── examples/ +│ ├── assess_accuracy.py +│ ├── compare_runs.py +│ ├── run_framework.py +│ └── run_tests.py +├── src/ +│ ├── testing/ +│ │ └── test_framework.py +│ └── utils/ +│ ├── Utils.py +│ ├── accuracy.py +│ ├── analysis.py +│ ├── comparison.py +│ ├── config.py +│ ├── logger.py +│ └── visualization.py +├── tests/ +│ ├── integration/ +│ │ └── test_api.py +│ └── unit/ +│ ├── test_accuracy.py +│ ├── test_analysis.py +│ └── test_comparison.py +├── main.py +├── README.md +└── requirements.txt +``` + +## Libraries Installed + +To run this code and utilize all the bonus features (like statistical analysis and data visualization), the following Python libraries were installed via `requirements.txt`: +- **`pytest`**: For running the automated unit and integration test suites. +- **`matplotlib`**: For rendering line charts of the collected ammeter data. +- **`pyyaml`**: For parsing the configuration-driven framework approach. + +## Usage Guide + +The framework is driven entirely by `config/config.yaml`. Before running any commands, ensure your desired sampling rates, durations, and output paths are set correctly in the configuration file. + +### 1. Start the Ammeter Emulators (Server) +Before running any tests, you must start the local ammeter emulators. This script reads `config.yaml` to dynamically bind the correct ports and commands. ```sh python3 main.py ``` +*(Leave this running in the background or in a separate terminal tab).* + +### 2. Run a Simple Sequential Test +To verify the framework is connected and operational, run a simple, sequential test sequence across all three ammeters. This will print a clean terminal summary without generating archives. +```sh +python3 examples/run_tests.py +``` + +### 3. Run a Production Automated Test & Visualization +To run a full test against a specific ammeter, generating a JSON archive and a Matplotlib visualization line-chart: +```sh +python3 examples/run_framework.py --ammeter greenlee +``` +*(Available options: `greenlee`, `entes`, `circutor`).* +The resulting JSON file and `.png` graph will automatically be saved into the `results/` directory. + +### 4. Assess Accuracy & Precision (Bonus) +To determine which ammeter is the most precise and accurate, this script executes concurrent, multi-threaded sampling across all three emulators simultaneously. It calculates the Ensemble Mean (consensus) and provides a clean terminal report highlighting the winners. +```sh +python3 examples/assess_accuracy.py +``` + +### 5. Compare Historical Runs +If you want to evaluate two historical JSON archives side-by-side, use the comparison utility. +```sh +python3 examples/compare_runs.py +``` +*(You will be prompted to paste the filenames of the two JSON files you wish to compare).* + +### 6. Enable Error Simulation / Chaos Mode (Bonus) +To simulate hardware faults and network drops, open `config/config.yaml` and set: +```yaml +testing: + error_simulation: true +``` +Restart `main.py`. The emulators will now randomly sleep beyond timeouts, return malformed garbage bytes, or abruptly drop connections. The `client.py` and test framework will gracefully catch these errors and auto-recover. --- @@ -81,7 +125,9 @@ When you execute a test run using the framework, it generates a comprehensive JS "median": 0.2876, "min": 0.0197, "max": 5.8526, - "stdev": 2.3266 + "stdev": 2.3266, + "cv_percentage": 157.0753, + "is_consistent": false }, "test_id": "a1b2c3d4-xxxx-xxxx-xxxx-xxxxxxxxxxxx", "timestamp": "20260506_145000", @@ -92,16 +138,24 @@ When you execute a test run using the framework, it generates a comprehensive JS --- -## Design Decisions & Bug Fixes +## Design Decisions & Bonus Features + +### 1. Configuration-Driven Testing (Bonus) & Main.py Refactor +- Unified the configuration to make `config/config.yaml` and the emulator classes the single source of truth. +- Refactored `main.py` to dynamically load port bindings and commands rather than hardcoding them. Added a robust verification loop that polls the sockets on startup instead of relying on an arbitrary `time.sleep()`. + +### 2. Error Simulation / Chaos Mode (Bonus) +- Added an `error_simulation` boolean toggle in the `testing` block of `config.yaml`. +- When enabled, `main.py` passes the `chaos_mode` flag to the ammeters, causing them to randomly inject hardware faults 10% of the time (e.g., sleeping beyond the client timeout, returning malformed byte strings, or abruptly closing the connection). +- The client connection logic (`Ammeters/client.py`) was refactored with a robust `try-except` block to catch `socket.timeout`, `ValueError`, and `ConnectionError` gracefully without crashing the active test. + +### 3. Accuracy Assessment & Concurrency (Bonus) (Issue #4) +- Extracted mathematical aggregation into a dedicated, unit-tested module `src/utils/accuracy.py`. +- Created an executable script `examples/assess_accuracy.py` utilizing Python's `concurrent.futures.ThreadPoolExecutor` to simultaneously fetch samples from all emulators, calculate the ensemble mean, and highlight the most precise and accurate devices in a formatted terminal report. -### 1. Emulator Communication Fix (PR #10) -**The Problem:** The initial `main.py` script failed to fetch data from the ammeter emulators. -**The Fix:** -- Discovered discrepancies between the documented ports/commands, `main.py`, and the actual `Ammeters/*_Ammeter.py` implementations. -- Unified the configuration to make `config/config.yaml` and the emulator classes the source of truth. -- Set the ammeters to listen sequentially on ports `5000` (Greenlee), `5001` (ENTES), and `5002` (CIRCUTOR). -- Updated `main.py` to send the correct byte strings (e.g., `b'MEASURE_CIRCUTOR -get_measurement'`). -- Added `socket.SO_REUSEADDR` to `base_ammeter.py` to prevent "Address already in use" errors during rapid test iterations. +### 4. Performance Consistency Evaluation (Bonus) +- Updated the core statistical payload in `src/utils/analysis.py` to compute the **Coefficient of Variation (CV %)** (`(stdev / mean) * 100`). +- Added a configurable threshold to evaluate a new boolean metric `is_consistent`, returning true only if the device's CV is below a 5.0% deviation baseline. ### 2. Unified Testing API (Issue #7) **The Problem:** The exam requires a unified interface capable of communicating consistently with multiple ammeter types. diff --git a/examples/assess_accuracy.py b/examples/assess_accuracy.py new file mode 100644 index 0000000..5f75eea --- /dev/null +++ b/examples/assess_accuracy.py @@ -0,0 +1,78 @@ +import sys +from pathlib import Path +import concurrent.futures + +# Ensure the parent directory is in the sys.path so we can import src +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from src.utils.logger import TestLogger +logger = TestLogger("AssessAccuracy").logger + +from src.testing.test_framework import AmmeterTestFramework # noqa: E402 +from src.utils.accuracy import evaluate_accuracy # noqa: E402 + +def run_ammeter_test(ammeter_type): + framework = AmmeterTestFramework() + + # Disable visualization to prevent matplotlib threading crashes (double free/corruption) + if 'analysis' in framework.config and 'visualization' in framework.config['analysis']: + framework.config['analysis']['visualization']['enabled'] = False + + # run_test handles sampling and returns a dictionary of results + return framework.run_test(ammeter_type) + +def main(): + logger.info("Starting Accuracy Assessment...") + ammeters = ['greenlee', 'entes', 'circutor'] + results = {} + + # 1. Concurrency: Run tests simultaneously + logger.info("Running tests concurrently for all ammeters...") + with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: + future_to_ammeter = {executor.submit(run_ammeter_test, a): a for a in ammeters} + for future in concurrent.futures.as_completed(future_to_ammeter): + ammeter_type = future_to_ammeter[future] + try: + data = future.result() + results[ammeter_type] = data + except Exception as exc: + logger.error(f"{ammeter_type} generated an exception: {exc}") + + try: + # Evaluate accuracy, data aggregation, relative accuracy calculation, scoring & identification + evaluation = evaluate_accuracy(results) + except ValueError as e: + logger.error(f"Error during accuracy evaluation: {e}. Exiting.") + sys.exit(1) + + # CLI Output + print("\n--- Accuracy Assessment Report ---") + print(f"Ensemble Mean (Consensus): {evaluation['ensemble_mean']:.6f} A\n") + + print(f"{'Ammeter':<12} | {'Mean (A)':<12} | {'Abs Error (A)':<15} | {'CV (%)':<10}") + print("-" * 57) + + most_accurate = evaluation['most_accurate'] + most_precise = evaluation['most_precise'] + + for ammeter_type, mets in evaluation['metrics'].items(): + mean_str = f"{mets['mean']:.6f}" + err_str = f"{mets['abs_error']:.6f}" + cv_str = f"{mets['cv_percentage']:.4f}" + + flags = [] + if ammeter_type == most_accurate: + flags.append("🏆 Most Accurate") + if ammeter_type == most_precise: + flags.append("🎯 Most Precise") + + flag_str = " ".join(flags) + + print(f"{ammeter_type.capitalize():<12} | {mean_str:<12} | {err_str:<15} | {cv_str:<10} {flag_str}") + + print("\nSummary:") + print(f"- The Most Accurate Ammeter is: **{most_accurate.capitalize()}**") + print(f"- The Most Precise/Reliable Ammeter is: **{most_precise.capitalize()}**") + +if __name__ == "__main__": + main() diff --git a/main.py b/main.py index 8653142..2bf4534 100644 --- a/main.py +++ b/main.py @@ -1,51 +1,76 @@ import threading import time +import socket from Ammeters.Circutor_Ammeter import CircutorAmmeter 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 +from src.utils.logger import TestLogger +logger = TestLogger("Main").logger -def run_greenlee_emulator(chaos_mode: bool = False): - greenlee = GreenleeAmmeter(5000, chaos_mode=chaos_mode) +def run_greenlee_emulator(port: int, chaos_mode: bool = False): + greenlee = GreenleeAmmeter(port, chaos_mode=chaos_mode) greenlee.start_server() -def run_entes_emulator(chaos_mode: bool = False): - entes = EntesAmmeter(5001, chaos_mode=chaos_mode) +def run_entes_emulator(port: int, chaos_mode: bool = False): + entes = EntesAmmeter(port, chaos_mode=chaos_mode) entes.start_server() -def run_circutor_emulator(chaos_mode: bool = False): - circutor = CircutorAmmeter(5002, chaos_mode=chaos_mode) +def run_circutor_emulator(port: int, chaos_mode: bool = False): + circutor = CircutorAmmeter(port, chaos_mode=chaos_mode) circutor.start_server() -def start_emulators(chaos_mode: bool = False): +def wait_for_port(port: int, timeout: int = 5): + """Wait until a port is bound and listening before proceeding.""" + start_time = time.time() + while time.time() - start_time < timeout: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + if s.connect_ex(('localhost', port)) == 0: + return True + time.sleep(0.1) + return False + +def start_emulators(config: dict): + chaos_mode = config.get("testing", {}).get("error_simulation", False) + + greenlee_cfg = config.get("ammeters", {}).get("greenlee", {}) + entes_cfg = config.get("ammeters", {}).get("entes", {}) + circutor_cfg = config.get("ammeters", {}).get("circutor", {}) + # Start each ammeter in a separate thread - 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() + threading.Thread(target=run_greenlee_emulator, args=(greenlee_cfg.get('port', 5000), chaos_mode), daemon=True).start() + threading.Thread(target=run_entes_emulator, args=(entes_cfg.get('port', 5001), chaos_mode), daemon=True).start() + threading.Thread(target=run_circutor_emulator, args=(circutor_cfg.get('port', 5002), 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) + # Robustly wait for the servers to start + ports_to_wait = [greenlee_cfg.get('port', 5000), entes_cfg.get('port', 5001), circutor_cfg.get('port', 5002)] + for port in ports_to_wait: + if not wait_for_port(port): + logger.warning(f"Timeout waiting for emulator on port {port} to start.") if __name__ == "__main__": 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) + logger.info(f"Starting ammeter emulators... (Chaos Mode: {chaos_mode})") + start_emulators(config) - # 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') + # Request an initial reading just to verify connection dynamically + logger.info("Verifying connections...") + for name, cfg in config.get("ammeters", {}).items(): + port = cfg.get("port") + command = cfg.get("command", "").encode("utf-8") + if port and command: + request_current_from_ammeter(port, command) - print("\nEmulators are running in the background. Press Ctrl+C to stop.") + logger.info("Emulators 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.") + logger.info("Shutting down emulators.") pass diff --git a/requirements.txt b/requirements.txt index 0e65b3d..9d607d8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,3 +4,4 @@ matplotlib>=3.4.0 seaborn>=0.11.0 pyyaml>=6.0 pandas>=1.3.0 +pytest>=8.0.0 diff --git a/src/utils/accuracy.py b/src/utils/accuracy.py new file mode 100644 index 0000000..ac3cd85 --- /dev/null +++ b/src/utils/accuracy.py @@ -0,0 +1,53 @@ +import statistics +from typing import Dict, Any, List + +def evaluate_accuracy(results: Dict[str, Dict[str, Any]]) -> Dict[str, Any]: + """ + Evaluates the relative accuracy and precision of multiple ammeters based on their test results. + + Args: + results: A dictionary where keys are ammeter types and values are test result dictionaries + (containing 'measurements' and optionally 'statistics'). + + Returns: + A dictionary containing the ensemble mean, individual metrics, and identified winners. + """ + if not results: + raise ValueError("Cannot evaluate accuracy with empty results.") + + # Data Aggregation + all_measurements: List[float] = [] + for data in results.values(): + measurements = data.get('measurements', []) + all_measurements.extend(measurements) + + if not all_measurements: + raise ValueError("No measurements found across any ammeter results.") + + ensemble_mean = statistics.mean(all_measurements) + + # Relative Accuracy Calculation + metrics: Dict[str, Dict[str, float]] = {} + for ammeter_type, data in results.items(): + stats = data.get('statistics', {}) + ammeter_mean = stats.get('mean', 0.0) + cv_percentage = stats.get('cv_percentage', float('inf')) + + abs_error = abs(ammeter_mean - ensemble_mean) + + metrics[ammeter_type] = { + 'mean': ammeter_mean, + 'abs_error': abs_error, + 'cv_percentage': cv_percentage + } + + # Scoring & Identification + most_accurate = min(metrics.keys(), key=lambda k: metrics[k]['abs_error']) + most_precise = min(metrics.keys(), key=lambda k: metrics[k]['cv_percentage']) + + return { + "ensemble_mean": ensemble_mean, + "metrics": metrics, + "most_accurate": most_accurate, + "most_precise": most_precise + } diff --git a/tests/integration/test_api.py b/tests/integration/test_api.py index fe898cf..1586baa 100644 --- a/tests/integration/test_api.py +++ b/tests/integration/test_api.py @@ -8,7 +8,9 @@ def ammeter_emulators(): """Starts the ammeter emulators in the background for integration tests.""" from main import start_emulators - start_emulators() + from src.utils.config import load_config + config = load_config("config/config.yaml") + start_emulators(config) yield # Daemon threads will terminate automatically when the pytest session completes diff --git a/tests/unit/test_accuracy.py b/tests/unit/test_accuracy.py new file mode 100644 index 0000000..cd0d698 --- /dev/null +++ b/tests/unit/test_accuracy.py @@ -0,0 +1,58 @@ +import pytest +from src.utils.accuracy import evaluate_accuracy + +def test_evaluate_accuracy_success(): + results = { + "device1": { + "measurements": [1.0, 2.0, 3.0], + "statistics": { + "mean": 2.0, + "cv_percentage": 50.0 + } + }, + "device2": { + "measurements": [2.5, 3.0, 3.5], + "statistics": { + "mean": 3.0, + "cv_percentage": 10.0 + } + }, + "device3": { + "measurements": [0.5, 1.0, 1.5], + "statistics": { + "mean": 1.0, + "cv_percentage": 100.0 + } + } + } + + evaluation = evaluate_accuracy(results) + + # All measurements: [1.0, 2.0, 3.0, 2.5, 3.0, 3.5, 0.5, 1.0, 1.5] + # Sum: 18.0, Count: 9, Ensemble Mean: 2.0 + assert evaluation["ensemble_mean"] == pytest.approx(2.0, rel=1e-5) + + # Abs Error = |Mean - Ensemble Mean| + # device1: |2.0 - 2.0| = 0.0 -> Most Accurate + # device2: |3.0 - 2.0| = 1.0 + # device3: |1.0 - 2.0| = 1.0 + assert evaluation["metrics"]["device1"]["abs_error"] == 0.0 + assert evaluation["metrics"]["device2"]["abs_error"] == 1.0 + assert evaluation["metrics"]["device3"]["abs_error"] == 1.0 + + assert evaluation["most_accurate"] == "device1" + + # Lowest CV is device2 (10.0) -> Most Precise + assert evaluation["most_precise"] == "device2" + +def test_evaluate_accuracy_empty_results(): + with pytest.raises(ValueError, match="Cannot evaluate accuracy with empty results"): + evaluate_accuracy({}) + +def test_evaluate_accuracy_no_measurements(): + results = { + "device1": {"measurements": []}, + "device2": {"measurements": []} + } + with pytest.raises(ValueError, match="No measurements found across any ammeter results"): + evaluate_accuracy(results)