start_streaming
Initiates real-time data acquisition from PicoScope oscilloscopes with configurable sampling intervals, buffer sizes, and capture duration for continuous or limited signal streaming.
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 |
|---|---|---|---|
| sample_interval_ns | Yes | ||
| buffer_size | No | ||
| auto_stop | No | ||
| max_samples | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- The MCP tool handler for 'start_streaming'. Decorated with @mcp.tool(), defines input schema via type hints, and contains the core logic (currently stubbed with TODO).
@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)Top-level registration call in the main server file that invokes register_acquisition_tools(mcp), thereby registering the 'start_streaming' tool among others.
register_acquisition_tools(mcp) - Supporting method in the device manager class intended for the actual streaming implementation, called potentially from 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