Update BioSPPy/mcp_output/mcp_plugin/mcp_service.py

BioSPPy/mcp_output/mcp_plugin/mcp_service.py

@@ -1,201 +1,160 @@
-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 an ECG signal and extract R-peaks and heart rate")
-def process_ecg(signal: List[float], sampling_rate: float = 1000.0) -> dict:
+@mcp.tool(name="process_ecg", description="Process ECG signals and extract features")
+def process_ecg(signal: list, sampling_rate: int) -> dict:
     """
-    Process a raw ECG signal and extract relevant signal features.
+    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.
 
-    :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.
+    Returns:
+    - dict: Processed ECG features including heart rate and R-peaks.
     """
     try:
-        signal_array = np.array(signal)
-        result = ecg.ecg(signal=signal_array, sampling_rate=sampling_rate, show=False, interactive=False)
+        from biosppy.signals import ecg
+        
+        output = ecg.ecg(signal=signal, sampling_rate=sampling_rate, show=False)
         
         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
+            "heart_rate": output[6],
+            "rpeaks": output[2].tolist()
         }
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
+        return {"success": False, "error": str(e)}
 
-
-@mcp.tool(name="process_eda", description="Process an EDA (electrodermal activity) signal")
-def process_eda(signal: List[float], sampling_rate: float = 1000.0) -> dict:
+@mcp.tool(name="process_resp", description="Process respiratory signals and extract features")
+def process_resp(signal: list, sampling_rate: int) -> dict:
     """
-    Process a raw EDA signal and extract relevant features.
+    Process respiratory signals and extract features.
+
+    Parameters:
+    - signal: List of respiratory signal values.
+    - sampling_rate: Sampling rate of the signal in Hz.
 
-    :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.
+    Returns:
+    - dict: Processed respiratory features.
     """
     try:
-        signal_array = np.array(signal)
-        result = eda.eda(signal=signal_array, sampling_rate=sampling_rate, show=False)
+        from biosppy.signals import resp
+        
+        output = resp.resp(signal=signal, sampling_rate=sampling_rate, show=False)
         
         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
+            "resp_rate": output[5],
+            "filtered_signal": output[1].tolist()
         }
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
+        return {"success": False, "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:
+@mcp.tool(name="process_eda", description="Process electrodermal activity (EDA) signals")
+def process_eda(signal: list, sampling_rate: int) -> dict:
     """
-    Process a raw EMG signal and extract relevant features.
+    Process electrodermal activity (EDA) signals and extract features.
+
+    Parameters:
+    - signal: List of EDA signal values.
+    - sampling_rate: Sampling rate of the signal in Hz.
 
-    :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.
+    Returns:
+    - dict: Processed EDA features.
     """
     try:
-        signal_array = np.array(signal)
-        result = emg.emg(signal=signal_array, sampling_rate=sampling_rate, show=False)
+        from biosppy.signals import eda
+        
+        output = eda.eda(signal=signal, sampling_rate=sampling_rate, show=False)
         
         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
+            "filtered_signal": output[1].tolist(),
+            "onsets": output[2].tolist()
         }
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
+        return {"success": False, "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:
+@mcp.tool(name="process_emg", description="Process electromyography (EMG) signals")
+def process_emg(signal: list, sampling_rate: int) -> dict:
     """
-    Process a raw respiration signal and extract relevant features.
+    Process electromyography (EMG) signals and extract features.
+
+    Parameters:
+    - signal: List of EMG signal values.
+    - sampling_rate: Sampling rate of the signal in Hz.
 
-    :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.
+    Returns:
+    - dict: Processed EMG features.
     """
     try:
-        signal_array = np.array(signal)
-        result = resp.resp(signal=signal_array, sampling_rate=sampling_rate, show=False)
+        from biosppy.signals import emg
+        
+        output = emg.emg(signal=signal, sampling_rate=sampling_rate, show=False)
         
         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
+            "filtered_signal": output[1].tolist()
         }
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
+        return {"success": False, "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:
+@mcp.tool(name="process_pcg", description="Process phonocardiogram (PCG) signals")
+def process_pcg(signal: list, sampling_rate: int) -> 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.
+    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.
     """
     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
-        )
+        from biosppy.signals import pcg
+        
+        output = pcg.pcg(signal=signal, sampling_rate=sampling_rate, show=False)
         
         return {
             "success": True,
-            "result": {
-                "filtered": filtered.tolist()[:100] if len(filtered) > 100 else filtered.tolist(),
-                "length": len(filtered),
-            },
-            "error": None
+            "filtered_signal": output[1].tolist()
         }
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
+        return {"success": False, "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:
+@mcp.tool(name="process_ecg_synthesis", description="Synthesize ECG signals")
+def process_ecg_synthesis(duration: float, sampling_rate: int) -> dict:
     """
-    Calculate heart rate from R-peak indices.
+    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.
 
-    :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.
+    Returns:
+    - dict: Synthesized ECG signal.
     """
     try:
-        rpeaks_array = np.array(rpeaks)
-        hr_idx, hr = tools.get_heart_rate(beats=rpeaks_array, sampling_rate=sampling_rate, smooth=True, size=3)
+        from biosppy.synthesizers import ecg
+        
+        signal = ecg.ecg(duration=duration, sampling_rate=sampling_rate)
         
         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
+            "synthesized_signal": signal.tolist()
         }
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
+        return {"success": False, "error": str(e)}
 
-def create_app():
+def create_app() -> FastMCP:
     """
     Create and return the FastMCP application instance.
 
-    :return: The FastMCP application instance.
+    Returns:
+    - FastMCP: The FastMCP application instance.
     """
-    return mcp
-    """Create and return FastMCP application instance"""
-    return mcp
-
-if __name__ == "__main__":
-    mcp.run(transport="http", host="0.0.0.0", port=8000)
+    return mcp