start_streaming
Initiate continuous data acquisition from PicoScope oscilloscopes by configuring sample intervals, buffer size, and capture duration for real-time signal analysis.
Instructions
Start streaming data acquisition.
Args: sample_interval_ns: Sample interval in nanoseconds. buffer_size: Size of streaming buffer. auto_stop: Whether to automatically stop after max_samples. max_samples: Maximum samples to capture (0 = continuous).
Returns: Dictionary containing streaming status and configuration.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| auto_stop | No | ||
| buffer_size | No | ||
| max_samples | No | ||
| sample_interval_ns | Yes |
Implementation Reference
- The primary handler function for the MCP 'start_streaming' tool. It is decorated with @mcp.tool() for automatic registration and handles the tool execution logic (currently a TODO stub).@mcp.tool() def start_streaming( sample_interval_ns: int, buffer_size: int = 100000, auto_stop: bool = False, max_samples: int = 0, ) -> dict[str, Any]: """Start streaming data acquisition. Args: sample_interval_ns: Sample interval in nanoseconds. buffer_size: Size of streaming buffer. auto_stop: Whether to automatically stop after max_samples. max_samples: Maximum samples to capture (0 = continuous). Returns: Dictionary containing streaming status and configuration. """ # TODO: Implement streaming mode return { "status": "not_implemented", "sample_interval_ns": sample_interval_ns, "buffer_size": buffer_size, }
- src/picoscope_mcp/server.py:17-17 (registration)Registration call to register_acquisition_tools(mcp), which defines and registers the start_streaming tool among others.register_acquisition_tools(mcp)
- Supporting method in device_manager for starting streaming on the hardware device, intended to be called by the tool handler.def start_streaming( self, sample_interval_ns: int, buffer_size: int ) -> bool: """Start streaming mode. Args: sample_interval_ns: Sample interval in nanoseconds. buffer_size: Buffer size for streaming. Returns: True if successful, False otherwise. """ if not self.is_connected(): return False # TODO: Implement streaming mode # This will need to set up buffers and call ps*RunStreaming return True