Add tests and CI

.github/workflows/ci.yml

@@ -0,0 +1,50 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+    paths:
+      - "**.py"
+      - "pyproject.toml"
+      - "requirements*.txt"
+      - ".github/workflows/ci.yml"
+  pull_request:
+    paths:
+      - "**.py"
+      - "pyproject.toml"
+      - "requirements*.txt"
+      - ".github/workflows/ci.yml"
+
+permissions:
+  contents: read
+
+jobs:
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.x"
+      - name: Install Ruff
+        run: python -m pip install ruff
+      - name: Lint with Ruff
+        run: python -m ruff check --output-format=github
+
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.10", "3.11", "3.12", "3.13"]
+    steps:
+      - uses: actions/checkout@v6
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: pip
+          cache-dependency-path: requirements-dev.txt
+      - name: Install test dependencies
+        run: pip install -r requirements-dev.txt
+      - name: Run unit tests
+        run: pytest -v

app.py

@@ -9,9 +9,10 @@
 import time
 import threading
 import requests
+from utils.actions import repair_json_response, transform_actions_to_receiver_format
 from utils.llm import LLMClient
 from utils.speech import VoiceCommandListener
-from utils.prompts import SYSTEM_PROMPT, PLANNING_PROMPT, CHECK_PROMPT
+from utils.prompts import SYSTEM_PROMPT
 
 RECEIVER_URL = os.getenv("RECEIVER_URL", "http://127.0.0.1:5001/execute")
 
@@ -20,229 +21,47 @@
 FRAME_RESIZE = (512, 384)
 JPEG_QUALITY = 60
 
-# EMS defaults
-EMS_AMPLITUDE = 60
-EMS_DURATION = 1.0
-EMS_FREQUENCY = 100
-EMS_PULSE_WIDTH = 1000
-
 
 def get_latest_frame() -> bytes:
     """
     Capture the latest frame from the local camera with retry logic.
     """
     max_retries = 3
-    
+
     for attempt in range(max_retries):
         try:
             cap = cv2.VideoCapture(CAMERA_INDEX)
-            
+
             if not cap.isOpened():
-                if attempt < max_retries - 1:
-                    print(f"[!] Camera not opened (attempt {attempt+1}/{max_retries}), retrying...")
-                    time.sleep(0.5)
-                    continue
-                else:
-                    raise Exception(f"Failed to open camera {CAMERA_INDEX} after {max_retries} attempts")
-
+                raise RuntimeError(f"Failed to open camera {CAMERA_INDEX}")
+
             ret, frame = cap.read()
             cap.release()
-            
+
             if not ret:
-                if attempt < max_retries - 1:
-                    print(f"[!] Failed to read frame (attempt {attempt+1}/{max_retries}), retrying...")
-                    time.sleep(0.5)
-                    continue
-                else:
-                    raise Exception(f"Failed to read frame from camera after {max_retries} attempts")
-
+                raise RuntimeError("Failed to read frame from camera")
+
             # Resize and encode with quality settings (matching vlm_test.py approach)
             resized = cv2.resize(frame, FRAME_RESIZE)
             encode_params = [cv2.IMWRITE_JPEG_QUALITY, JPEG_QUALITY]
             _, buffer = cv2.imencode('.jpg', resized, encode_params)
             return buffer.tobytes()
-        
+
         except Exception as e:
             if attempt == max_retries - 1:
                 print(f"[!] Error getting frame from camera: {e}")
                 raise
+            print(f"[!] {e} (attempt {attempt + 1}/{max_retries}), retrying...")
             time.sleep(0.5)
 
     raise RuntimeError("Failed to capture frame after retries")
 
 
 
-def action_to_finger_mapping(action: str) -> str:
-    """
-    Map action names to receiver finger/command codes.
-
-    Map actions to relay target names expected by firmware.
-
-    Supported relay targets:
-    - wrist_left, wrist_right, thumb, index, middle, ring, pinky
-    - x = reset/all off
-    """
-    mapping = {
-        "clench_hand": "x",      # reset/end sequence
-        "close_index": "index",
-        "close_middle": "middle",
-        "close_pinky": "pinky",
-        "close_thumb": "thumb",
-        "close_ring": "ring",
-        "wrist_left": "wrist_left",
-        "wrist_right": "wrist_right",
-        # These are not relay-compatible and will be skipped:
-        # "biceps_flex": requires different command type
-        # "lean_left": GVS command, not relay
-        # "lean_right": GVS command, not relay
-    }
-    return mapping.get(action, "x")
-
-
-
-
-def transform_actions_to_receiver_format(claude_response: dict) -> dict:
-    """
-    Transform Claude's response format to receiver.py's timestamped format.
-
-    Handles both formats:
-    INPUT (numeric keys):
-    {
-      "1": [["close_middle", 1.0], ["clench_hand", 0.5]],
-      "2": [["close_pinky", 2.0]]
-    }
-
-    INPUT (sequence keys):
-    {
-      'sequence_1': [['close_middle', 1.0], ['clench_hand', 0.5]],
-      'sequence_2': [['close_pinky', 2.0]]
-    }
-
-    OUTPUT (for receiver.py):
-    {
-      "0": [
-        {"type": "RELAY", "finger": "m"},
-        {"type": "EMS", "channel": 1, "amplitude": 60, "duration": 1.0, "frequency": 100}
-      ],
-      "1.0": [
-        {"type": "RELAY", "finger": "x"},
-        {"type": "EMS", "channel": 1, "amplitude": 60, "duration": 0.5, "frequency": 100}
-      ],
-      "1.5": [
-        {"type": "RELAY", "finger": "p"},
-        {"type": "EMS", "channel": 1, "amplitude": 60, "duration": 2.0, "frequency": 100}
-      ]
-    }
-
-    Timing logic:
-    - Each action starts at cumulative_time (sum of all previous durations)
-    - Duration in the action is how long the EMS stimulation lasts
-    - All supported actions select one relay target then stimulate EMS channel 1
-    - Unsupported actions (biceps, lean) are logged but not sent
-    """
-    receiver_format = {}
-    current_time = 0.0
-
-    # Determine key format and sort accordingly
-    numeric_keys = [k for k in claude_response.keys() if k.isdigit()]
-    sequence_keys = [k for k in claude_response.keys() if k.startswith('sequence_')]
-
-    if numeric_keys:
-        # Sort by numeric value
-        sorted_keys = sorted(numeric_keys, key=lambda x: int(x))
-    elif sequence_keys:
-        # Sort by sequence number
-        sorted_keys = sorted(sequence_keys, key=lambda x: int(x.split('_')[1]))
-    else:
-        # Unknown format, use as-is
-        sorted_keys = list(claude_response.keys())
-
-    for key in sorted_keys:
-        actions = claude_response[key]
-
-        for action_name, duration in actions:
-            # Map action to finger code
-            finger_code = action_to_finger_mapping(action_name)
-
-            # Create timestamped entry key
-            time_key = str(current_time)
-
-            # Skip unsupported actions (biceps, lean, etc.)
-            if action_name in ["biceps_flex", "lean_left", "lean_right"]:
-                print(f"[!] Skipping unsupported action: {action_name}")
-                current_time += float(duration) + 1.0
-                continue
-
-            if time_key not in receiver_format:
-                receiver_format[time_key] = []
-
-            # All supported actions: RELAY select first, then EMS on channel 1
-            receiver_format[time_key].append({
-                "type": "RELAY",
-                "finger": finger_code
-            })
-            receiver_format[time_key].append({
-                "type": "EMS",
-                "channel": 1,
-                "amplitude": EMS_AMPLITUDE,
-                "duration": float(duration),
-                "frequency": EMS_FREQUENCY,
-                "pulse_width": EMS_PULSE_WIDTH
-            })
-
-            # Move to next action time (current duration + 1 second buffer for relay to open)
-            current_time += float(duration) + 1.0
-
-    # Always append "x" command (disable all fingers) at the end
-    final_time = str(current_time)
-    receiver_format[final_time] = [{
-        "type": "RELAY",
-        "finger": "x"
-    }]
-
-    return receiver_format
-
-
-def repair_json_response(raw_text: str) -> str:
-    """
-    Repair common JSON formatting issues from Claude.
-    Handles unquoted numeric keys like: 1: [...] -> "1": [...]
-    """
-    import re
-
-    # Extract content between curly braces
-    text = raw_text.strip()
-
-    # Remove markdown code fences if present
-    if "```" in text:
-        parts = text.split("```")
-        if len(parts) >= 3:
-            inner = parts[1]
-            if inner.startswith("json"):
-                inner = inner[4:]
-            text = inner.strip()
-
-    # Find the JSON object content (between { and })
-    start = text.find('{')
-    end = text.rfind('}') + 1
-
-    if start == -1 or end == 0:
-        raise ValueError("No JSON object found")
-
-    json_content = text[start:end]
-
-    # Fix unquoted numeric keys: change `1:` to `"1":`
-    # Pattern: word boundary, one or more digits, colon
-    json_content = re.sub(r'(\n\s*)(\d+):', r'\1"\2":', json_content)
-
-    return json_content
-
-
-
 def on_command_ready(command: str):
     """Callback when voice command is ready"""
 
-    print(f"\n[*] Getting latest frame...")
+    print("\n[*] Getting latest frame...")
     frame_bytes = get_latest_frame()
 
     # Create prompt combining frame content and voice command
@@ -286,14 +105,14 @@ def on_command_ready(command: str):
 def execute_motor_commands(receiver_payload: dict):
     """Send motor command sequence to receiver.py"""
     try:
-        print(f"[*] Sending motor commands to receiver...")
+        print("[*] Sending motor commands to receiver...")
         response = requests.post(RECEIVER_URL, json=receiver_payload, timeout=10)
 
         if response.status_code == 200:
             print(f"[✓] Receiver acknowledged (HTTP {response.status_code})")
             try:
                 print(f"    Response: {response.json()}")
-            except:
+            except ValueError:
                 print(f"    Response body: {response.text}")
         else:
             print(f"[!] Receiver returned HTTP {response.status_code}")

manual_control_app.py

@@ -1,6 +1,6 @@
 from PyQt5.QtWidgets import (QApplication, QWidget, QLabel, QPushButton, QVBoxLayout, 
                             QSlider, QLineEdit, QHBoxLayout, QRadioButton, 
-                            QButtonGroup, QGridLayout, QGroupBox, QSizePolicy, QMessageBox)
+                            QButtonGroup, QGridLayout, QGroupBox, QMessageBox)
 from PyQt5.QtCore import Qt
 
 from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas
@@ -504,12 +504,12 @@ def save_settings(self):
             filename = f"settings/human_operator_cfg_{timestamp}.txt"
 
             with open(filename, "w") as f:
-                f.write(f"SYSTEM: HUMAN OPERATOR\n")
-                f.write(f"MODE: EMS\n")
+                f.write("SYSTEM: HUMAN OPERATOR\n")
+                f.write("MODE: EMS\n")
                 f.write(f"CHANNEL: {self.get_channel()}\n")
                 f.write(f"FREQUENCY: {self.sliders['FREQUENCY']['slider'].value()} Hz\n")
                 f.write(f"PULSE WIDTH: {self.sliders['PULSE WIDTH']['slider'].value()} μs\n")
-                f.write(f"POLARITY: BIPHASIC\n")  
+                f.write("POLARITY: BIPHASIC\n")  
                 f.write(f"AMPLITUDE: {self.sliders['AMPLITUDE']['slider'].value()} {self.sliders['AMPLITUDE']['unit']}\n")
                 f.write(f"DURATION: {self.sliders['DURATION']['slider'].value()} ms\n")
 

pyproject.toml

@@ -0,0 +1,13 @@
+[tool.pytest.ini_options]
+# importlib mode is pytest's recommended import mode for new projects:
+# https://docs.pytest.org/en/stable/explanation/goodpractices.html
+addopts = ["--import-mode=importlib"]
+testpaths = ["tests"]
+pythonpath = ["."]
+
+[tool.ruff]
+target-version = "py310"
+
+[tool.ruff.lint.per-file-ignores]
+# Legacy PyQt GUI relies on a star import from the external hcint_estim module.
+"manual_control_app.py" = ["F403", "F405"]

requirements-dev.txt

@@ -0,0 +1,6 @@
+# Dependencies needed to run the test suite and linter (no camera/audio/GUI
+# stack required — tests run fully headless in simulation mode).
+flask
+pyserial
+pytest
+ruff