Record JLS

Instructions

Implementation Reference

• src/joulescope_mcp/service.py:372-417 (handler)

  The core implementation of the record_jls tool. Opens a driver session, selects a device, configures auto range, creates a Record writer, writes raw JS220 samples to a JLS v2 file for the specified duration, then returns metadata about the recording.

  def record_jls(
      self,
      output_path: str,
      duration_s: float,
      device_path: str | None = None,
      frequency_hz: int | None = None,
      signals: str = "current,voltage,power",
      note: str | None = None,
      overwrite: bool = False,
  ) -> dict[str, Any]:
      if duration_s <= 0:
          raise JoulescopeMcpError("duration_s must be greater than 0")
      if duration_s > self._limits.max_duration_s:
          raise JoulescopeMcpError(f"duration_s must be <= {self._limits.max_duration_s}")
      path = Path(output_path).expanduser().resolve()
      path.parent.mkdir(parents=True, exist_ok=True)
      if path.exists() and not overwrite:
          raise JoulescopeMcpError(f"Output path already exists. Set overwrite=true to replace: {path}")
  
      with self._driver_session() as driver:
          device = self._select_device(driver, device_path=device_path)
          driver.open(device, mode="defaults")
          writer = None
          try:
              if frequency_hz is not None:
                  driver.publish(f"{device}/h/fs", int(frequency_hz))
              driver.publish(f"{device}/s/i/range/mode", "auto")
              driver.publish(f"{device}/s/v/range/mode", "auto")
              writer = self._record_factory(driver, [device], signals)
              user_data = [] if note is None else [[0, note]]
              writer.open(str(path), user_data=user_data)
              stop_at = time.monotonic() + float(duration_s)
              while time.monotonic() < stop_at:
                  time.sleep(min(0.05, stop_at - time.monotonic()))
          finally:
              if writer is not None:
                  writer.close()
              driver.close(device)
      return {
          "device_path": device,
          "output_path": str(path),
          "duration_s": float(duration_s),
          "signals": signals,
          "frequency_hz": frequency_hz,
          "overwrite": overwrite,
      }

• src/joulescope_mcp/server.py:137-139 (schema)

  The MCP tool registration with FastMCP, defining input parameters/output schema via type annotations: output_path (str), duration_s (float), device_path (str|None), frequency_hz (int|None), signals (str), note (str|None), overwrite (bool). Decorated with @mcp.tool annotations including file_write_tool for destructive hint.

          raise _tool_error(exc) from exc
  
  @mcp.tool(

• src/joulescope_mcp/server.py:145-164 (registration)

  The delegate function registered as an MCP tool. It calls service.record_jls(...) and wraps errors in ToolError.

  def record_jls(
      output_path: str,
      duration_s: float,
      device_path: str | None = None,
      frequency_hz: int | None = None,
      signals: str = "current,voltage,power",
      note: str | None = None,
      overwrite: bool = False,
  ) -> dict[str, Any]:
      try:
          return service.record_jls(
              output_path=output_path,
              duration_s=duration_s,
              device_path=device_path,
              frequency_hz=frequency_hz,
              signals=signals,
              note=note,
              overwrite=overwrite,
          )
      except JoulescopeMcpError as exc:

• src/joulescope_mcp/service.py:48-51 (helper)

  The _default_record_factory helper that creates a pyjoulescope_driver.Record object used inside record_jls to write JLS v2 files.

  def _default_record_factory(driver: Any, devices: list[str], signals: str) -> Any:
      if Record is None:
          raise JoulescopeMcpError("pyjoulescope_driver.Record is not available")
      return Record(driver, devices, signals)