From 7f6f5f660bd58f47c2df1bfd4da7a2960de36e67 Mon Sep 17 00:00:00 2001 From: Ava Barron Date: Thu, 16 Jul 2026 08:35:49 -0400 Subject: [PATCH] Add tests and CI. Quality now enforces itself. --- .github/workflows/ci.yml | 50 ++++++++++ app.py | 209 +++------------------------------------ manual_control_app.py | 8 +- pyproject.toml | 13 +++ requirements-dev.txt | 6 ++ tests/test_actions.py | 134 +++++++++++++++++++++++++ tests/test_receiver.py | 94 ++++++++++++++++++ utils/actions.py | 177 +++++++++++++++++++++++++++++++++ utils/ball_demo.py | 12 ++- utils/receiver.py | 120 ++++++++++++---------- 10 files changed, 566 insertions(+), 257 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 pyproject.toml create mode 100644 requirements-dev.txt create mode 100644 tests/test_actions.py create mode 100644 tests/test_receiver.py create mode 100644 utils/actions.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..53a0900 --- /dev/null +++ b/.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 diff --git a/app.py b/app.py index acc8663..8e2d0f5 100644 --- a/app.py +++ b/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}") diff --git a/manual_control_app.py b/manual_control_app.py index 78eb530..8f64839 100644 --- a/manual_control_app.py +++ b/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") diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..629c261 --- /dev/null +++ b/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"] diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..64b8dfc --- /dev/null +++ b/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 diff --git a/tests/test_actions.py b/tests/test_actions.py new file mode 100644 index 0000000..ee346ef --- /dev/null +++ b/tests/test_actions.py @@ -0,0 +1,134 @@ +"""Unit tests for the pure action-plan transformation logic in utils/actions.py.""" + +import json + +import pytest + +from utils.actions import ( + EMS_AMPLITUDE, + EMS_FREQUENCY, + EMS_PULSE_WIDTH, + action_to_finger_mapping, + repair_json_response, + transform_actions_to_receiver_format, +) + + +class TestActionToFingerMapping: + def test_known_actions_map_to_relay_targets(self): + assert action_to_finger_mapping("close_index") == "index" + assert action_to_finger_mapping("close_middle") == "middle" + assert action_to_finger_mapping("close_ring") == "ring" + assert action_to_finger_mapping("close_pinky") == "pinky" + assert action_to_finger_mapping("close_thumb") == "thumb" + assert action_to_finger_mapping("wrist_left") == "wrist_left" + assert action_to_finger_mapping("wrist_right") == "wrist_right" + + def test_clench_hand_maps_to_reset(self): + assert action_to_finger_mapping("clench_hand") == "x" + + def test_unknown_action_defaults_to_reset(self): + assert action_to_finger_mapping("do_a_backflip") == "x" + + +class TestTransformActionsToReceiverFormat: + def test_single_action_emits_relay_then_ems_then_final_reset(self): + result = transform_actions_to_receiver_format( + {"1": [["close_index", 1.0]]} + ) + + assert result["0.0"] == [ + {"type": "RELAY", "finger": "index"}, + { + "type": "EMS", + "channel": 1, + "amplitude": EMS_AMPLITUDE, + "duration": 1.0, + "frequency": EMS_FREQUENCY, + "pulse_width": EMS_PULSE_WIDTH, + }, + ] + # 1.0s duration + 1.0s relay buffer -> trailing all-off at 2.0 + assert result["2.0"] == [{"type": "RELAY", "finger": "x"}] + assert set(result.keys()) == {"0.0", "2.0"} + + def test_cumulative_timing_across_actions(self): + result = transform_actions_to_receiver_format( + {"1": [["close_index", 1.0], ["close_middle", 2.0]]} + ) + + # Second action starts after first duration + 1s buffer. + assert result["2.0"][0] == {"type": "RELAY", "finger": "middle"} + # Final reset at 2.0 + 2.0 + 1.0 = 5.0 + assert result["5.0"] == [{"type": "RELAY", "finger": "x"}] + + def test_numeric_keys_sorted_numerically_not_lexically(self): + result = transform_actions_to_receiver_format( + { + "10": [["close_middle", 1.0]], + "2": [["close_index", 1.0]], + } + ) + + # Key "2" must execute first even though "10" < "2" lexically. + assert result["0.0"][0]["finger"] == "index" + assert result["2.0"][0]["finger"] == "middle" + + def test_sequence_keys_supported(self): + result = transform_actions_to_receiver_format( + { + "sequence_2": [["close_middle", 1.0]], + "sequence_1": [["close_index", 1.0]], + } + ) + + assert result["0.0"][0]["finger"] == "index" + assert result["2.0"][0]["finger"] == "middle" + + def test_non_step_keys_like_plan_are_ignored(self): + result = transform_actions_to_receiver_format( + { + "plan": "close the index finger", + "1": [["close_index", 1.0]], + } + ) + + fingers = [cmd["finger"] for cmds in result.values() for cmd in cmds if cmd["type"] == "RELAY"] + assert fingers == ["index", "x"] + + def test_unsupported_actions_skipped_but_still_advance_time(self): + result = transform_actions_to_receiver_format( + {"1": [["biceps_flex", 2.0], ["close_index", 1.0]]} + ) + + # biceps_flex is dropped entirely, but its 2.0 + 1.0 window still elapses. + assert "0.0" not in result + assert result["3.0"][0] == {"type": "RELAY", "finger": "index"} + assert result["5.0"] == [{"type": "RELAY", "finger": "x"}] + + def test_empty_plan_still_ends_with_all_off(self): + result = transform_actions_to_receiver_format({}) + + assert result == {"0.0": [{"type": "RELAY", "finger": "x"}]} + + +class TestRepairJsonResponse: + def test_plain_json_passes_through(self): + raw = '{"1": [["close_index", 1.0]]}' + assert json.loads(repair_json_response(raw)) == {"1": [["close_index", 1.0]]} + + def test_strips_markdown_code_fences(self): + raw = 'Here you go:\n```json\n{"1": [["close_index", 1.0]]}\n```\nDone.' + assert json.loads(repair_json_response(raw)) == {"1": [["close_index", 1.0]]} + + def test_quotes_unquoted_numeric_keys(self): + raw = '{\n 1: [["close_index", 1.0]]\n}' + assert json.loads(repair_json_response(raw)) == {"1": [["close_index", 1.0]]} + + def test_extracts_object_from_surrounding_prose(self): + raw = 'Sure! {"plan": "wave"} hope that helps' + assert json.loads(repair_json_response(raw)) == {"plan": "wave"} + + def test_raises_when_no_json_object_present(self): + with pytest.raises(ValueError): + repair_json_response("no json here at all") diff --git a/tests/test_receiver.py b/tests/test_receiver.py new file mode 100644 index 0000000..ee9cac5 --- /dev/null +++ b/tests/test_receiver.py @@ -0,0 +1,94 @@ +"""Smoke tests for the Flask hardware gateway in simulation (relay-only) mode. + +receiver.py configures its hardware at import time from environment variables, +so the module is imported once per test session with the environment pinned to +relay mode and a nonexistent serial port. That guarantees SIMULATED mode on any +machine — including a developer laptop with a real relay plugged in. +""" + +import importlib +import os +import sys + +import pytest + +EMS_COMMAND = { + "type": "EMS", + "channel": 1, + "amplitude": 60, + "duration": 0.1, + "frequency": 100, + "pulse_width": 1000, +} + + +@pytest.fixture(scope="session") +def receiver_module(): + os.environ["HARDWARE_MODE"] = "relay" + os.environ["RELAY_PORT"] = "/dev/nonexistent-test-port" + os.environ.pop("STIM_PORT", None) + os.environ.pop("ENABLE_STIM", None) + + sys.modules.pop("utils.receiver", None) + module = importlib.import_module("utils.receiver") + yield module + sys.modules.pop("utils.receiver", None) + + +@pytest.fixture() +def client(receiver_module): + receiver_module.app.config["TESTING"] = True + with receiver_module.app.test_client() as test_client: + yield test_client + + +class TestHealth: + def test_health_reports_relay_mode_simulated(self, client): + response = client.get("/health") + payload = response.get_json() + + assert response.status_code == 200 + assert payload["status"] == "ready" + assert payload["hardware_mode"] == "relay" + assert payload["relay_hardware_connected"] is False + assert payload["stim_hardware_connected"] is False + assert payload["stim_port"] == "SIMULATED" + + +class TestExecute: + def test_relay_command_executes_in_simulation(self, client): + response = client.post( + "/execute", + json={"0.0": [{"type": "RELAY", "finger": "index"}]}, + ) + payload = response.get_json() + + assert response.status_code == 200 + assert payload["status"] == "executed" + assert payload["hardware_mode"]["relay"] == "SIMULATED" + + def test_ems_command_is_skipped_in_relay_mode(self, client): + response = client.post( + "/execute", + json={"0.0": [{"type": "RELAY", "finger": "middle"}, EMS_COMMAND]}, + ) + + assert response.status_code == 200 + assert response.get_json()["status"] == "executed" + + def test_full_transformed_payload_shape_is_accepted(self, client): + # The exact shape app.py produces via transform_actions_to_receiver_format. + payload = { + "0.0": [{"type": "RELAY", "finger": "index"}, EMS_COMMAND], + "0.1": [{"type": "RELAY", "finger": "x"}], + } + response = client.post("/execute", json=payload) + + assert response.status_code == 200 + assert response.get_json()["status"] == "executed" + + def test_malformed_payload_returns_error_not_crash(self, client): + response = client.post("/execute", json=["not", "a", "dict"]) + + assert response.status_code == 500 + assert "error" in response.get_json() diff --git a/utils/actions.py b/utils/actions.py new file mode 100644 index 0000000..fe405b8 --- /dev/null +++ b/utils/actions.py @@ -0,0 +1,177 @@ +#!/usr/bin/env python3 +""" +Pure transformation logic for turning Claude action plans into receiver payloads. + +Kept free of hardware/UI dependencies (cv2, serial, audio) so it can be unit +tested without a camera, microphone, or relay board attached. +""" + +import re + +# EMS defaults +EMS_AMPLITUDE = 60 +EMS_DURATION = 1.0 +EMS_FREQUENCY = 100 +EMS_PULSE_WIDTH = 1000 + + +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": [...] + """ + # 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 diff --git a/utils/ball_demo.py b/utils/ball_demo.py index ca0dd3d..fe18e62 100644 --- a/utils/ball_demo.py +++ b/utils/ball_demo.py @@ -8,13 +8,14 @@ import cv2 import numpy as np +import os import requests import time -import json from collections import deque # --- Configuration --- -RECEIVER_URL = "https://amsterdam-river-lease-toolbox.trycloudflare.com/execute" +# Same env-var convention as app.py. +RECEIVER_URL = os.getenv("RECEIVER_URL", "http://127.0.0.1:5001/execute") EMS_COMMAND = { "0": [ @@ -28,9 +29,6 @@ ] } -response = requests.post(RECEIVER_URL, json=RELAY_COMMAND, timeout=5) -print(f"[HTTP] Sent RELAY command to {RECEIVER_URL} -> {response.status_code}") - # HSV range for bright orange # Ball 1 Settings # BALL_LOW = np.array([165, 125, 0]) @@ -64,6 +62,10 @@ def detect_approach(area_buffer): def main(): + # Select the middle-finger relay path before the demo starts. + response = requests.post(RECEIVER_URL, json=RELAY_COMMAND, timeout=5) + print(f"[HTTP] Sent RELAY command to {RECEIVER_URL} -> {response.status_code}") + # The front facing (OV5640) defaults to 0 camera_index_1 = 0 diff --git a/utils/receiver.py b/utils/receiver.py index 77bed37..f309930 100644 --- a/utils/receiver.py +++ b/utils/receiver.py @@ -18,32 +18,57 @@ # } # Note: EMS uses channel 1. Relay selects which electrode path is active. # Relay commands can be: wrist_left, wrist_right, thumb, index, middle, ring, pinky, x + + +class SerialDevice: + """Shared connect/close/send-or-simulate behavior for serial hardware.""" + + label = "device" + + def __init__(self, port=None, baudrate=115200, timeout=1): + self.port = port + self.ser = None + self.error = None + if port: + try: + self.ser = serial.Serial(port, baudrate, timeout=timeout) + self._on_connect() + except Exception as e: + self.error = str(e) + print(f"❌ {self.label.upper()} SERIAL CONNECTION ERROR: {e}") + + def _on_connect(self): + print(f"✅ Connected to {self.label} on {self.port}") + + def _format_command(self, cmd): + return cmd.strip() + + def _after_send(self): + pass + + def close(self): + if self.ser: + self.ser.close() + + def send_command(self, cmd): + if self.ser and self.ser.is_open: + self.ser.write(cmd.encode('utf-8')) + self.ser.flush() + print(f"✅ [REAL HARDWARE SENT] {self._format_command(cmd)}") + self._after_send() + else: + print(f"⚠️ [SIMULATION MODE] {self._format_command(cmd)}") + if self.error: + print(f" Error: {self.error}") + + # Import the existing stimulator class try: from hcint_estim import HCIntEstim # type: ignore except ImportError: - class HCIntEstim: - def __init__(self, port=None, baudrate=115200, timeout=1): - self.port = port - self.ser = None - self.error = None - if port: - try: - self.ser = serial.Serial(port, baudrate, timeout=timeout) - print(f"✅ Connected to stimulator on {port}") - except Exception as e: - self.error = str(e) - print(f"❌ STIMULATOR SERIAL CONNECTION ERROR: {e}") - def close(self): - if self.ser: self.ser.close() - def send_command(self, cmd): - if self.ser and self.ser.is_open: - self.ser.write(cmd.encode()) - self.ser.flush() - print(f"✅ [REAL HARDWARE SENT] {cmd.strip()}") - else: - print(f"⚠️ [SIMULATION MODE] {cmd.strip()}") - if self.error: print(f" Error: {self.error}") + class HCIntEstim(SerialDevice): + label = "stimulator" + def stim_ems(self, channel, amplitude, freq, pulse_width, duration): self.send_command(f"ems,{channel},{amplitude},{freq},{pulse_width},{duration}\n") def stim_gvs(self, channel, amplitude, polarity, duration): @@ -51,22 +76,25 @@ def stim_gvs(self, channel, amplitude, polarity, duration): def stim_et(self, channel, amplitude, polarity, freq, pulse_width, duration): self.send_command(f"et,{channel},{amplitude},{polarity},{freq},{pulse_width},{duration}\n") -# New Relay Controller Class -class RelayController: + +class RelayController(SerialDevice): + label = "relay MCU" + def __init__(self, port=None, baudrate=115200, timeout=0.1): - self.port = port - self.ser = None - self.error = None - if port: - try: - self.ser = serial.Serial(port, baudrate, timeout=timeout) - time.sleep(1.5) # Let the Arduino reset and boot - print(f"✅ Connected to relay MCU on {port}") - for line in self._read_available_lines(max_wait=0.8): - print(f"↩️ [RELAY BOOT] {line}") - except Exception as e: - self.error = str(e) - print(f"❌ RELAY SERIAL CONNECTION ERROR: {e}") + super().__init__(port, baudrate, timeout) + + def _on_connect(self): + time.sleep(1.5) # Let the Arduino reset and boot + print(f"✅ Connected to relay MCU on {self.port}") + for line in self._read_available_lines(max_wait=0.8): + print(f"↩️ [RELAY BOOT] {line}") + + def _format_command(self, cmd): + return f"Relay: {cmd.strip()}" + + def _after_send(self): + for line in self._read_available_lines(max_wait=0.25): + print(f"↩️ [RELAY RX] {line}") def _read_available_lines(self, max_wait=0.25): if not (self.ser and self.ser.is_open): @@ -92,20 +120,6 @@ def _read_available_lines(self, max_wait=0.25): return lines - def close(self): - if self.ser: self.ser.close() - - def send_command(self, cmd): - if self.ser and self.ser.is_open: - self.ser.write(cmd.encode('utf-8')) - self.ser.flush() - print(f"✅ [REAL HARDWARE SENT] Relay: {cmd.strip()}") - for line in self._read_available_lines(max_wait=0.25): - print(f"↩️ [RELAY RX] {line}") - else: - print(f"⚠️ [SIMULATION MODE] Relay: {cmd.strip()}") - if self.error: print(f" Error: {self.error}") - app = Flask(__name__) @@ -191,7 +205,7 @@ def execute_sequence(): # Sort sequence keys try: sorted_keys = sorted(data.keys(), key=lambda x: float(x)) - except: + except (TypeError, ValueError): sorted_keys = data.keys() start_time = time.monotonic() @@ -245,7 +259,7 @@ def execute_sequence(): return jsonify({"error": str(e)}), 500 if __name__ == '__main__': - print(f"🚀 Receiver starting on port 5001...") + print("🚀 Receiver starting on port 5001...") print(f"🛠️ Stimulator Port: {stim_port or 'NONE'}") print(f"🛠️ Relay Port: {relay_port or 'NONE'}") app.run(host='0.0.0.0', port=5001) \ No newline at end of file