feat: add testing unified api

.github/workflows/pull-request.yaml

@@ -0,0 +1,23 @@
+name: Ammeter Test Framework CI
+
+on: pull_request
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v6
+
+    - name: Set up Python 3.10
+      uses: actions/setup-python@v6
+      with:
+        python-version: '3.10'
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+
+    - name: Run Integration Tests
+      run: python tests/integration/test_api.py

.gitignore

@@ -216,3 +216,6 @@ __marimo__/
 
 # Streamlit
 .streamlit/secrets.toml
+
+# Gemini AI configuration
+.gemini/

Ammeters/base_ammeter.py

@@ -16,6 +16,8 @@ def start_server(self):
         The server will run indefinitely, handling one client request at a time.
         """
         with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+            # Allow the socket to be reused immediately after the program exits
+            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}")

Ammeters/client.py

@@ -1,13 +1,16 @@
 from socket import socket, AF_INET, SOCK_STREAM
+from typing import Optional
 
-
-def request_current_from_ammeter(port: int, command: bytes):
+def request_current_from_ammeter(port: int, command: bytes) -> Optional[float]:
     with socket(AF_INET, SOCK_STREAM) as s:
         s.connect(('localhost', port))
         s.sendall(command)
         data = s.recv(1024)
         if data:
-            print(f"Received current measurement from port {port}: {data.decode('utf-8')} A")
+            current = data.decode('utf-8')
+            print(f"Received current measurement from port {port}: {current} A")
+            return float(current)
         else:
             print("No data received.")
+            return None
 

README.md

@@ -50,5 +50,26 @@ This project provides emulators for different types of ammeters: Greenlee, ENTES
 
 To start the ammeter emulators and request current measurements, run the `main.py` script:
 ```sh
-python main.py
-```
+python3 main.py
+```
+
+---
+
+## Design Decisions & Bug Fixes
+
+### 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.
+
+### 2. Unified Testing API (Current)
+**The Problem:** The exam requires a unified interface capable of communicating consistently with multiple ammeter types.
+**The Design:**
+- Implemented `AmmeterTestFramework` in `src/testing/test_framework.py`.
+- The framework acts as an abstraction layer; users simply call `get_single_reading('greenlee')` without needing to manage raw sockets, ports, or byte commands.
+- The framework dynamically reads the required connection parameters from `config/config.yaml`.
+- The base `client.py` was updated to decode and return standard Python `float` types rather than printing to stdout, enabling programmatic data aggregation.

config/config.yaml

@@ -5,15 +5,15 @@ testing:
     sampling_frequency_hz: NULL
 
 ammeters:
-#  greenlee:
-#    port: 5000
-#    command: "MEASURE_GREENLEE -get_measurement"
-#  entes:
-#    port: 5001
-#    command: "MEASURE_ENTES -get_data"
-#  circutor:
-#    port: 5002
-#    command: "MEASURE_CIRCUTOR -get_measurement"
+  greenlee:
+    port: 5000
+    command: "MEASURE_GREENLEE -get_measurement"
+  entes:
+    port: 5001
+    command: "MEASURE_ENTES -get_data"
+  circutor:
+    port: 5002
+    command: "MEASURE_CIRCUTOR -get_measurement"
 
 analysis:
   statistical_metrics:

main.py

@@ -19,17 +19,18 @@ def run_circutor_emulator():
     circutor = CircutorAmmeter(5002)
     circutor.start_server()
 
-if __name__ == "__main__":
+def start_emulators():
     # 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()
-
-    # This section is commented out because it shouldn't work.
-    # Read the README.md file as well as the source code if you need, and fix the problem.
-
+
     # 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__":
+    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

src/testing/test_framework.py

@@ -1,11 +1,26 @@
 
 
+from typing import Dict, Optional
 from ..utils.config import load_config
+from Ammeters.client import request_current_from_ammeter
 
 
 class AmmeterTestFramework:
     def __init__(self, config_path: str = "config/config.yaml"):
         self.config = load_config(config_path)
 
+    def get_single_reading(self, ammeter_type: str) -> Optional[float]:
+        """
+        Fetches a single current reading from the specified ammeter.
+        """
+        if ammeter_type not in self.config.get('ammeters', {}):
+            raise ValueError(f"Unknown ammeter type: {ammeter_type}")
+
+        ammeter_config = self.config['ammeters'][ammeter_type]
+        port = ammeter_config['port']
+        command = ammeter_config['command'].encode('utf-8')
+
+        return request_current_from_ammeter(port, command)
+
     def run_test(self, ammeter_type: str) -> Dict:
         pass

tests/integration/test_api.py

@@ -0,0 +1,54 @@
+import sys
+import os
+
+# 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 main import start_emulators
+from src.testing.test_framework import AmmeterTestFramework
+
+if __name__ == "__main__":
+    print("Starting ammeter emulators in the background...")
+    start_emulators()
+    print("Emulators started. Testing the new API...\n")
+
+    # Initialize your new framework
+    # Note: We need to pass the absolute path to the config file so it works from any directory
+    config_path = os.path.join(project_root, 'config', 'config.yaml')
+    framework = AmmeterTestFramework(config_path=config_path)
+
+    failures = 0
+
+    # Test reading from each valid type
+    for ammeter in ['greenlee', 'entes', 'circutor']:
+        print(f"--- Requesting reading from {ammeter.upper()} ---")
+        try:
+            value = framework.get_single_reading(ammeter)
+            if value is None:
+                print(f"❌ Failed: API returned None instead of a value for {ammeter}\n")
+                failures += 1
+            else:
+                print(f"✅ Success! API returned type: {type(value).__name__}, value: {value}\n")
+        except Exception as e:
+            print(f"❌ Failed to fetch {ammeter}: {e}\n")
+            failures += 1
+
+    # Test reading from an invalid type (Expected to fail)
+    print(f"--- Requesting reading from UNKNOWN (Expected to raise ValueError) ---")
+    try:
+        framework.get_single_reading('unknown')
+        print(f"❌ Failed: Expected ValueError, but reading succeeded.\n")
+        failures += 1
+    except ValueError as e:
+        print(f"✅ Success! API correctly raised ValueError for unknown ammeter.\n")
+    except Exception as e:
+        print(f"❌ Failed: Expected ValueError, but got {type(e).__name__}: {e}\n")
+        failures += 1
+
+    if failures > 0:
+        print(f"Integration tests failed with {failures} error(s).")
+        sys.exit(1)
+    else:
+        print("All integration tests passed successfully!")
+        sys.exit(0)