Update BioSPPy/mcp_output/mcp_plugin/mcp_service.py

BioSPPy/mcp_output/mcp_plugin/mcp_service.py

@@ -1,103 +1,199 @@
 import os
 import sys
+from typing import List, Optional
 
 source_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "source")
 if source_path not in sys.path:
     sys.path.insert(0, source_path)
 
 from fastmcp import FastMCP
+import numpy as np
 
-from setup import UploadCommand
-from docs.conf import Mock
+# Import BioSPPy modules
+from biosppy.signals import ecg, eda, emg, resp, ppg, eeg, tools
 
-mcp = FastMCP("unknown_service")
+mcp = FastMCP("biosppy_service")
 
 
-@mcp.tool(name="uploadcommand", description="UploadCommand class")
-def uploadcommand(*args, **kwargs):
-    """UploadCommand class"""
+@mcp.tool(name="process_ecg", description="Process an ECG signal and extract R-peaks and heart rate")
+def process_ecg(signal: List[float], sampling_rate: float = 1000.0) -> dict:
+    """
+    Process a raw ECG signal and extract relevant signal features.
+
+    :param signal: Raw ECG signal as a list of float values.
+    :param sampling_rate: Sampling frequency in Hz (default: 1000.0).
+    :return: Dictionary containing filtered signal, R-peaks, and heart rate.
+    """
     try:
-        if UploadCommand is None:
-            return {"success": False, "result": None, "error": "Class UploadCommand is not available, path may need adjustment"}
-        
-        # MCP parameter type conversion
-        converted_args = []
-        converted_kwargs = kwargs.copy()
-        
-        # Handle position argument type conversion
-        for arg in args:
-            if isinstance(arg, str):
-                # Try to convert to numeric type
-                try:
-                    if '.' in arg:
-                        converted_args.append(float(arg))
-                    else:
-                        converted_args.append(int(arg))
-                except ValueError:
-                    converted_args.append(arg)
-            else:
-                converted_args.append(arg)
-        
-        # Handle keyword argument type conversion
-        for key, value in converted_kwargs.items():
-            if isinstance(value, str):
-                try:
-                    if '.' in value:
-                        converted_kwargs[key] = float(value)
-                    else:
-                        converted_kwargs[key] = int(value)
-                except ValueError:
-                    pass
+        signal_array = np.array(signal)
+        result = ecg.ecg(signal=signal_array, sampling_rate=sampling_rate, show=False, interactive=False)
         
-        instance = UploadCommand(*converted_args, **converted_kwargs)
-        return {"success": True, "result": str(instance), "error": None}
+        return {
+            "success": True,
+            "result": {
+                "rpeaks": result['rpeaks'].tolist(),
+                "heart_rate": result['heart_rate'].tolist() if len(result['heart_rate']) > 0 else [],
+                "heart_rate_ts": result['heart_rate_ts'].tolist() if len(result['heart_rate_ts']) > 0 else [],
+                "num_beats": len(result['rpeaks']),
+            },
+            "error": None
+        }
     except Exception as e:
         return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="mock", description="Mock class")
-def mock(*args, **kwargs):
-    """Mock class"""
+
+@mcp.tool(name="process_eda", description="Process an EDA (electrodermal activity) signal")
+def process_eda(signal: List[float], sampling_rate: float = 1000.0) -> dict:
+    """
+    Process a raw EDA signal and extract relevant features.
+
+    :param signal: Raw EDA signal as a list of float values.
+    :param sampling_rate: Sampling frequency in Hz (default: 1000.0).
+    :return: Dictionary containing filtered signal and skin conductance responses.
+    """
     try:
-        if Mock is None:
-            return {"success": False, "result": None, "error": "Class Mock is not available, path may need adjustment"}
+        signal_array = np.array(signal)
+        result = eda.eda(signal=signal_array, sampling_rate=sampling_rate, show=False)
         
-        # MCP parameter type conversion
-        converted_args = []
-        converted_kwargs = kwargs.copy()
+        return {
+            "success": True,
+            "result": {
+                "filtered": result['filtered'].tolist()[:100] if len(result['filtered']) > 100 else result['filtered'].tolist(),
+                "onsets": result['onsets'].tolist() if hasattr(result['onsets'], 'tolist') else list(result['onsets']),
+                "peaks": result['peaks'].tolist() if hasattr(result['peaks'], 'tolist') else list(result['peaks']),
+                "amplitudes": result['amplitudes'].tolist() if hasattr(result['amplitudes'], 'tolist') else list(result['amplitudes']),
+            },
+            "error": None
+        }
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="process_emg", description="Process an EMG (electromyogram) signal")
+def process_emg(signal: List[float], sampling_rate: float = 1000.0) -> dict:
+    """
+    Process a raw EMG signal and extract relevant features.
+
+    :param signal: Raw EMG signal as a list of float values.
+    :param sampling_rate: Sampling frequency in Hz (default: 1000.0).
+    :return: Dictionary containing filtered signal and onsets.
+    """
+    try:
+        signal_array = np.array(signal)
+        result = emg.emg(signal=signal_array, sampling_rate=sampling_rate, show=False)
         
-        # Handle position argument type conversion
-        for arg in args:
-            if isinstance(arg, str):
-                # Try to convert to numeric type
-                try:
-                    if '.' in arg:
-                        converted_args.append(float(arg))
-                    else:
-                        converted_args.append(int(arg))
-                except ValueError:
-                    converted_args.append(arg)
-            else:
-                converted_args.append(arg)
+        return {
+            "success": True,
+            "result": {
+                "filtered": result['filtered'].tolist()[:100] if len(result['filtered']) > 100 else result['filtered'].tolist(),
+                "onsets": result['onsets'].tolist() if hasattr(result['onsets'], 'tolist') else list(result['onsets']),
+            },
+            "error": None
+        }
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="process_resp", description="Process a respiration signal")
+def process_resp(signal: List[float], sampling_rate: float = 1000.0) -> dict:
+    """
+    Process a raw respiration signal and extract relevant features.
+
+    :param signal: Raw respiration signal as a list of float values.
+    :param sampling_rate: Sampling frequency in Hz (default: 1000.0).
+    :return: Dictionary containing filtered signal and respiration rate.
+    """
+    try:
+        signal_array = np.array(signal)
+        result = resp.resp(signal=signal_array, sampling_rate=sampling_rate, show=False)
         
-        # Handle keyword argument type conversion
-        for key, value in converted_kwargs.items():
-            if isinstance(value, str):
-                try:
-                    if '.' in value:
-                        converted_kwargs[key] = float(value)
-                    else:
-                        converted_kwargs[key] = int(value)
-                except ValueError:
-                    pass
+        return {
+            "success": True,
+            "result": {
+                "filtered": result['filtered'].tolist()[:100] if len(result['filtered']) > 100 else result['filtered'].tolist(),
+                "zeros": result['zeros'].tolist() if hasattr(result['zeros'], 'tolist') else list(result['zeros']),
+                "resp_rate": result['resp_rate'].tolist() if hasattr(result['resp_rate'], 'tolist') else list(result['resp_rate']),
+            },
+            "error": None
+        }
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="filter_signal", description="Apply a digital filter to a signal")
+def filter_signal(signal: List[float], sampling_rate: float = 1000.0, 
+                  ftype: str = "butter", band: str = "lowpass", 
+                  frequency: float = 45.0, order: int = 4) -> dict:
+    """
+    Apply a digital filter to a signal.
+
+    :param signal: Input signal as a list of float values.
+    :param sampling_rate: Sampling frequency in Hz (default: 1000.0).
+    :param ftype: Filter type - 'butter', 'cheby1', 'cheby2', 'ellip', 'bessel', 'FIR' (default: 'butter').
+    :param band: Band type - 'lowpass', 'highpass', 'bandpass', 'bandstop' (default: 'lowpass').
+    :param frequency: Cutoff frequency in Hz (default: 45.0).
+    :param order: Filter order (default: 4).
+    :return: Dictionary containing filtered signal.
+    """
+    try:
+        signal_array = np.array(signal)
+        filtered, _, _ = tools.filter_signal(
+            signal=signal_array,
+            ftype=ftype,
+            band=band,
+            frequency=frequency,
+            order=order,
+            sampling_rate=sampling_rate
+        )
         
-        instance = Mock(*converted_args, **converted_kwargs)
-        return {"success": True, "result": str(instance), "error": None}
+        return {
+            "success": True,
+            "result": {
+                "filtered": filtered.tolist()[:100] if len(filtered) > 100 else filtered.tolist(),
+                "length": len(filtered),
+            },
+            "error": None
+        }
     except Exception as e:
         return {"success": False, "result": None, "error": str(e)}
 
 
+@mcp.tool(name="get_heart_rate", description="Calculate heart rate from R-peak indices")
+def get_heart_rate(rpeaks: List[int], sampling_rate: float = 1000.0) -> dict:
+    """
+    Calculate heart rate from R-peak indices.
+
+    :param rpeaks: List of R-peak indices.
+    :param sampling_rate: Sampling frequency in Hz (default: 1000.0).
+    :return: Dictionary containing heart rate values and statistics.
+    """
+    try:
+        rpeaks_array = np.array(rpeaks)
+        hr_idx, hr = tools.get_heart_rate(beats=rpeaks_array, sampling_rate=sampling_rate, smooth=True, size=3)
+        
+        return {
+            "success": True,
+            "result": {
+                "heart_rate": hr.tolist(),
+                "heart_rate_indices": hr_idx.tolist(),
+                "mean_hr": float(np.mean(hr)) if len(hr) > 0 else None,
+                "std_hr": float(np.std(hr)) if len(hr) > 0 else None,
+                "min_hr": float(np.min(hr)) if len(hr) > 0 else None,
+                "max_hr": float(np.max(hr)) if len(hr) > 0 else None,
+            },
+            "error": None
+        }
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
 
 def create_app():
+    """
+    Create and return the FastMCP application instance.
+
+    :return: The FastMCP application instance.
+    """
+    return mcp
     """Create and return FastMCP application instance"""
     return mcp