Update BioSPPy/mcp_output/mcp_plugin/mcp_service.py

BioSPPy/mcp_output/mcp_plugin/mcp_service.py

@@ -1,160 +1,201 @@
+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
+
+# Import BioSPPy modules
+from biosppy.signals import ecg, eda, emg, resp, ppg, eeg, tools
 
-# Create the FastMCP service application
 mcp = FastMCP("biosppy_service")
 
-@mcp.tool(name="process_ecg", description="Process ECG signals and extract features")
-def process_ecg(signal: list, sampling_rate: int) -> dict:
-    """
-    Process ECG signals and extract features such as heart rate and R-peaks.
 
-    Parameters:
-    - signal: List of ECG signal values.
-    - sampling_rate: Sampling rate of the signal in Hz.
+@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.
 
-    Returns:
-    - dict: Processed ECG features including heart rate and R-peaks.
+    :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:
-        from biosppy.signals import ecg
-        
-        output = ecg.ecg(signal=signal, sampling_rate=sampling_rate, show=False)
+        signal_array = np.array(signal)
+        result = ecg.ecg(signal=signal_array, sampling_rate=sampling_rate, show=False, interactive=False)
         
         return {
             "success": True,
-            "heart_rate": output[6],
-            "rpeaks": output[2].tolist()
+            "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, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="process_resp", description="Process respiratory signals and extract features")
-def process_resp(signal: list, sampling_rate: int) -> dict:
-    """
-    Process respiratory signals and extract features.
 
-    Parameters:
-    - signal: List of respiratory signal values.
-    - sampling_rate: Sampling rate of the signal in Hz.
+@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.
 
-    Returns:
-    - dict: Processed respiratory 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:
-        from biosppy.signals import resp
-        
-        output = resp.resp(signal=signal, sampling_rate=sampling_rate, show=False)
+        signal_array = np.array(signal)
+        result = eda.eda(signal=signal_array, sampling_rate=sampling_rate, show=False)
         
         return {
             "success": True,
-            "resp_rate": output[5],
-            "filtered_signal": output[1].tolist()
+            "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, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="process_eda", description="Process electrodermal activity (EDA) signals")
-def process_eda(signal: list, sampling_rate: int) -> dict:
-    """
-    Process electrodermal activity (EDA) signals and extract features.
 
-    Parameters:
-    - signal: List of EDA signal values.
-    - sampling_rate: Sampling rate of the signal in Hz.
+@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.
 
-    Returns:
-    - dict: Processed EDA 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:
-        from biosppy.signals import eda
-        
-        output = eda.eda(signal=signal, sampling_rate=sampling_rate, show=False)
+        signal_array = np.array(signal)
+        result = emg.emg(signal=signal_array, sampling_rate=sampling_rate, show=False)
         
         return {
             "success": True,
-            "filtered_signal": output[1].tolist(),
-            "onsets": output[2].tolist()
+            "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, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="process_emg", description="Process electromyography (EMG) signals")
-def process_emg(signal: list, sampling_rate: int) -> dict:
-    """
-    Process electromyography (EMG) signals and extract features.
 
-    Parameters:
-    - signal: List of EMG signal values.
-    - sampling_rate: Sampling rate of the signal in Hz.
+@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.
 
-    Returns:
-    - dict: Processed EMG 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:
-        from biosppy.signals import emg
-        
-        output = emg.emg(signal=signal, sampling_rate=sampling_rate, show=False)
+        signal_array = np.array(signal)
+        result = resp.resp(signal=signal_array, sampling_rate=sampling_rate, show=False)
         
         return {
             "success": True,
-            "filtered_signal": output[1].tolist()
+            "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, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="process_pcg", description="Process phonocardiogram (PCG) signals")
-def process_pcg(signal: list, sampling_rate: int) -> dict:
-    """
-    Process phonocardiogram (PCG) signals and extract features.
-
-    Parameters:
-    - signal: List of PCG signal values.
-    - sampling_rate: Sampling rate of the signal in Hz.
 
-    Returns:
-    - dict: Processed PCG features.
+@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:
-        from biosppy.signals import pcg
-        
-        output = pcg.pcg(signal=signal, sampling_rate=sampling_rate, show=False)
+        signal_array = np.array(signal)
+        filtered, _, _ = tools.filter_signal(
+            signal=signal_array,
+            ftype=ftype,
+            band=band,
+            frequency=frequency,
+            order=order,
+            sampling_rate=sampling_rate
+        )
         
         return {
             "success": True,
-            "filtered_signal": output[1].tolist()
+            "result": {
+                "filtered": filtered.tolist()[:100] if len(filtered) > 100 else filtered.tolist(),
+                "length": len(filtered),
+            },
+            "error": None
         }
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="process_ecg_synthesis", description="Synthesize ECG signals")
-def process_ecg_synthesis(duration: float, sampling_rate: int) -> dict:
-    """
-    Synthesize ECG signals for a given duration and sampling rate.
 
-    Parameters:
-    - duration: Duration of the synthesized signal in seconds.
-    - sampling_rate: Sampling rate of the signal in Hz.
+@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.
 
-    Returns:
-    - dict: Synthesized ECG signal.
+    :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:
-        from biosppy.synthesizers import ecg
-        
-        signal = ecg.ecg(duration=duration, sampling_rate=sampling_rate)
+        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,
-            "synthesized_signal": signal.tolist()
+            "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, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
+
 
-def create_app() -> FastMCP:
+def create_app():
     """
     Create and return the FastMCP application instance.
 
-    Returns:
-    - FastMCP: The FastMCP application instance.
+    :return: The FastMCP application instance.
     """
-    return mcp
+    return mcp
+    """Create and return FastMCP application instance"""
+    return mcp
+
+if __name__ == "__main__":
+    mcp.run(transport="http", host="0.0.0.0", port=8000)