PicoScope MCP Server

Instructions

Implementation Reference

• src/picoscope_mcp/tools/acquisition.py:74-125 (handler)

  MCP tool handler for 'capture_block'. Decorated with @mcp.tool(), calls device_manager.capture_block, formats and returns the captured data.

  @mcp.tool()
  def capture_block(
      pre_trigger_samples: int = 1000, post_trigger_samples: int = 1000
  ) -> dict[str, Any]:
      """Capture a single block of data.
  
      Args:
          pre_trigger_samples: Number of samples before trigger.
          post_trigger_samples: Number of samples after trigger.
  
      Returns:
          Dictionary containing captured waveform data for all enabled channels.
      """
      try:
          if not device_manager.is_connected():
              return {
                  "status": "error",
                  "error": "No device connected",
              }
  
          # Capture block
          result = device_manager.capture_block(pre_trigger_samples, post_trigger_samples)
  
          if result is None:
              return {
                  "status": "error",
                  "error": "Failed to capture data",
              }
  
          # Format the response
          channels_data = {}
          for channel, data in result.items():
              channels_data[channel] = {
                  "time_values": data.time_values,
                  "voltage_values": data.voltage_values,
                  "sample_interval_ns": data.sample_interval_ns,
                  "num_samples": data.num_samples,
              }
  
          return {
              "status": "success",
              "total_samples": pre_trigger_samples + post_trigger_samples,
              "pre_trigger_samples": pre_trigger_samples,
              "post_trigger_samples": post_trigger_samples,
              "channels": channels_data,
          }
  
      except Exception as e:
          return {
              "status": "error",
              "error": str(e),
          }

• src/picoscope_mcp/server.py:17-17 (registration)

  Registration of acquisition tools, which includes the capture_block tool, by calling register_acquisition_tools on the MCP server instance.

  register_acquisition_tools(mcp)

• src/picoscope_mcp/device_manager.py:339-483 (helper)

  Low-level helper method in PicoScopeManager that performs the actual block capture using PicoSDK API calls.

  def capture_block(
      self, pre_trigger: int, post_trigger: int
  ) -> Optional[dict[str, CaptureData]]:
      """Capture a block of data.
  
      Args:
          pre_trigger: Number of samples before trigger.
          post_trigger: Number of samples after trigger.
  
      Returns:
          Dictionary of channel data or None if failed.
      """
      if not self.is_connected():
          return None
  
      try:
          total_samples = pre_trigger + post_trigger
  
          # Set up buffers for enabled channels
          buffers = {}
          channel_map = {
              "A": ps.PS5000A_CHANNEL["PS5000A_CHANNEL_A"],
              "B": ps.PS5000A_CHANNEL["PS5000A_CHANNEL_B"],
              "C": ps.PS5000A_CHANNEL["PS5000A_CHANNEL_C"],
              "D": ps.PS5000A_CHANNEL["PS5000A_CHANNEL_D"],
          }
  
          for ch_name, ch_config in self.channel_configs.items():
              if ch_config.enabled:
                  # Create buffer
                  buffer = (ctypes.c_int16 * total_samples)()
                  buffers[ch_name] = buffer
  
                  # Set data buffer
                  self.status[f"setDataBuffer{ch_name}"] = ps.ps5000aSetDataBuffer(
                      self.chandle,
                      channel_map[ch_name],
                      ctypes.byref(buffer),
                      total_samples,
                      0,  # Segment index
                      ps.PS5000A_RATIO_MODE["PS5000A_RATIO_MODE_NONE"],
                  )
                  assert_pico_ok(self.status[f"setDataBuffer{ch_name}"])
  
          # Get timebase - using timebase 0 for fastest sampling
          timebase = 0
          time_interval_ns = ctypes.c_float()
          max_samples = ctypes.c_int32()
  
          self.status["getTimebase"] = ps.ps5000aGetTimebase2(
              self.chandle,
              timebase,
              total_samples,
              ctypes.byref(time_interval_ns),
              ctypes.byref(max_samples),
              0,  # Segment index
          )
  
          # If timebase 0 doesn't work, try higher values
          while self.status["getTimebase"] != 0 and timebase < 100:
              timebase += 1
              self.status["getTimebase"] = ps.ps5000aGetTimebase2(
                  self.chandle,
                  timebase,
                  total_samples,
                  ctypes.byref(time_interval_ns),
                  ctypes.byref(max_samples),
                  0,
              )
  
          assert_pico_ok(self.status["getTimebase"])
  
          # Run block capture
          self.status["runBlock"] = ps.ps5000aRunBlock(
              self.chandle,
              pre_trigger,
              post_trigger,
              timebase,
              None,  # Time indisposed (not used)
              0,  # Segment index
              None,  # Callback (not used, we'll poll)
              None,  # Callback parameter
          )
          assert_pico_ok(self.status["runBlock"])
  
          # Wait for capture to complete
          ready = ctypes.c_int16(0)
          check = ctypes.c_int16(0)
          while ready.value == check.value:
              self.status["isReady"] = ps.ps5000aIsReady(self.chandle, ctypes.byref(ready))
  
          # Get values
          overflow = ctypes.c_int16()
          c_total_samples = ctypes.c_int32(total_samples)
  
          self.status["getValues"] = ps.ps5000aGetValues(
              self.chandle,
              0,  # Start index
              ctypes.byref(c_total_samples),
              1,  # Downsampling ratio
              ps.PS5000A_RATIO_MODE["PS5000A_RATIO_MODE_NONE"],
              0,  # Segment index
              ctypes.byref(overflow),
          )
          assert_pico_ok(self.status["getValues"])
  
          # Convert ADC values to mV and create CaptureData objects
          result = {}
          time_values = np.arange(total_samples) * (time_interval_ns.value / 1e9)  # Convert to seconds
  
          for ch_name, buffer in buffers.items():
              ch_config = self.channel_configs[ch_name]
  
              # Get voltage range
              range_map = {
                  0.02: "PS5000A_20MV", 0.05: "PS5000A_50MV", 0.1: "PS5000A_100MV",
                  0.2: "PS5000A_200MV", 0.5: "PS5000A_500MV", 1.0: "PS5000A_1V",
                  2.0: "PS5000A_2V", 5.0: "PS5000A_5V", 10.0: "PS5000A_10V",
                  20.0: "PS5000A_20V",
              }
              closest_range = min(range_map.keys(), key=lambda x: abs(x - ch_config.voltage_range))
              voltage_range = ps.PS5000A_RANGE[range_map[closest_range]]
  
              # Convert to numpy array
              adc_array = np.array(buffer)
  
              # Convert ADC to mV
              voltage_mv = adc2mV(
                  adc_array,
                  voltage_range,
                  self.device_info.max_adc_value if self.device_info else 32767,
              )
  
              result[ch_name] = CaptureData(
                  channel=ch_name,
                  time_values=time_values.tolist(),
                  voltage_values=voltage_mv.tolist(),
                  sample_interval_ns=int(time_interval_ns.value),
                  num_samples=total_samples,
              )
  
          return result
  
      except Exception as e:
          return None