sigrok: csv and vcd parsing

Author: mangelajo

packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/__init__.py

@@ -1,5 +1,11 @@
-from jumpstarter_driver_sigrok.common import CaptureConfig, CaptureResult, DecoderConfig
+from jumpstarter_driver_sigrok.common import (
+    CaptureConfig,
+    CaptureResult,
+    DecoderConfig,
+    OutputFormat,
+    Sample,
+)
 from jumpstarter_driver_sigrok.driver import Sigrok
 
-__all__ = ["Sigrok", "CaptureConfig", "CaptureResult", "DecoderConfig"]
+__all__ = ["Sigrok", "CaptureConfig", "CaptureResult", "DecoderConfig", "OutputFormat", "Sample"]
 

packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/common.py

@@ -5,6 +5,27 @@
 from pydantic import BaseModel, Field
 
 
+class OutputFormat:
+    """Constants for sigrok output formats."""
+    CSV = "csv"
+    BITS = "bits"
+    ASCII = "ascii"
+    BINARY = "binary"
+    SRZIP = "srzip"
+    VCD = "vcd"
+
+    @classmethod
+    def all(cls) -> list[str]:
+        return [cls.CSV, cls.BITS, cls.ASCII, cls.BINARY, cls.SRZIP, cls.VCD]
+
+
+class Sample(BaseModel):
+    """A single sample with timing information."""
+    sample: int  # Sample index
+    time_ns: int  # Time in nanoseconds
+    values: dict[str, int | float]  # Channel values (digital: 0/1, analog: voltage)
+
+
 class DecoderConfig(BaseModel):
     """Protocol decoder configuration (real-time during capture)."""
 
@@ -47,3 +68,59 @@ def data(self) -> bytes:
         from base64 import b64decode
         return b64decode(self.data_b64)
 
+    def decode(self) -> list[Sample] | dict[str, list[int]] | str:
+        """Parse captured data based on output format.
+
+        Returns:
+            - CSV format: list[Sample] with timing and all values per sample
+            - VCD format: list[Sample] with timing and only changed values
+            - Bits format: dict[str, list[int]] with channel→bit sequences
+            - ASCII format: str with ASCII art visualization
+            - Other formats: raises NotImplementedError (use .data for raw bytes)
+
+        Raises:
+            NotImplementedError: For binary/srzip formats (use .data property)
+        """
+        if self.output_format == OutputFormat.CSV:
+            from .csv import parse_csv
+            samples_data = parse_csv(self.data, self.sample_rate)
+            return [Sample.model_validate(s) for s in samples_data]
+        elif self.output_format == OutputFormat.VCD:
+            from .vcd import parse_vcd
+            samples_data = parse_vcd(self.data, self.sample_rate)
+            return [Sample.model_validate(s) for s in samples_data]
+        elif self.output_format == OutputFormat.BITS:
+            return self._parse_bits()
+        elif self.output_format == OutputFormat.ASCII:
+            return self.data.decode("utf-8")
+        else:
+            raise NotImplementedError(
+                f"Parsing not implemented for {self.output_format} format. "
+                f"Use .data property to get raw bytes."
+            )
+
+    def _parse_bits(self) -> dict[str, list[int]]:
+        """Parse bits format to dict of channel→bit sequences."""
+        text = self.data.decode("utf-8")
+        lines = [line.strip() for line in text.strip().split("\n") if line.strip()]
+
+        # bits format is just columns of 0/1
+        # TODO: Need to determine channel mapping from somewhere
+        # For now, return as generic numbered channels
+        result: dict[str, list[int]] = {}
+
+        for line in lines:
+            # Each line might be space/comma separated bits
+            bits = [int(b) for b in line if b in "01"]
+            if not result:
+                # Initialize channels
+                for i, bit in enumerate(bits):
+                    result[f"CH{i}"] = [bit]
+            else:
+                # Append to existing channels
+                for i, bit in enumerate(bits):
+                    if f"CH{i}" in result:
+                        result[f"CH{i}"].append(bit)
+
+        return result
+

packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/csv.py

@@ -0,0 +1,141 @@
+"""CSV format parser for sigrok captures."""
+
+from __future__ import annotations
+
+import csv
+
+
+def parse_csv(data: bytes, sample_rate: str) -> list[dict]:
+    """Parse CSV format to list of samples with timing.
+
+    Args:
+        data: Raw CSV data as bytes
+        sample_rate: Sample rate string (e.g., "100kHz", "1MHz")
+
+    Returns:
+        List of dicts with keys: sample, time_ns, values
+    """
+    text = data.decode("utf-8")
+    lines = text.strip().split("\n")
+
+    # Parse sample rate for timing calculation
+    sample_rate_hz = _parse_sample_rate_hz(sample_rate)
+    time_step_ns = int(1_000_000_000.0 / sample_rate_hz)
+
+    # Skip comment lines and analog preview lines (format: "A0: -10.0000 V DC")
+    # The actual data starts after a header row with types like "logic,logic,V DC,V DC"
+    data_lines = _extract_csv_data_lines(lines)
+
+    if not data_lines or len(data_lines) < 2:
+        return []
+
+    # Parse the CSV data
+    reader = csv.reader(data_lines)
+
+    # First row is types (logic, V DC, etc.) - use for channel name inference
+    types_row = next(reader)
+
+    # Get channel names from types
+    channel_names = _infer_channel_names(types_row)
+
+    # Parse data rows
+    samples: list[dict] = []
+    for idx, row in enumerate(reader):
+        values = _parse_csv_row(channel_names, row)
+        samples.append({
+            "sample": idx,
+            "time_ns": idx * time_step_ns,
+            "values": values,
+        })
+
+    return samples
+
+
+def _parse_sample_rate_hz(sample_rate: str) -> float:
+    """Parse sample rate string to Hz."""
+    rate = sample_rate.strip().upper()
+    multipliers = {"K": 1e3, "M": 1e6, "G": 1e9}
+
+    for suffix, mult in multipliers.items():
+        if rate.endswith(f"{suffix}HZ"):
+            return float(rate[:-3]) * mult
+        elif rate.endswith(suffix):
+            return float(rate[:-1]) * mult
+
+    # Assume Hz if no suffix
+    return float(rate.rstrip("HZ"))
+
+
+def _extract_csv_data_lines(lines: list[str]) -> list[str]:
+    """Extract actual CSV data lines, skipping comments and analog preview lines."""
+    data_lines = []
+
+    for _i, line in enumerate(lines):
+        line = line.strip()
+        # Skip comment lines
+        if line.startswith(";"):
+            continue
+        # Skip analog preview lines (contain colon, not CSV comma-separated)
+        if ":" in line and "," not in line:
+            continue
+        # This is CSV data
+        data_lines.append(line)
+
+    return data_lines
+
+
+def _infer_channel_names(types_row: list[str]) -> list[str]:
+    """Infer channel names from CSV type header row.
+
+    Args:
+        types_row: List of type strings like ["logic", "logic", "V DC", "V DC"]
+
+    Returns:
+        List of channel names like ["D0", "D1", "A0", "A1"]
+    """
+    channel_names = []
+    digital_count = 0
+    analog_count = 0
+
+    for type_str in types_row:
+        type_lower = type_str.lower()
+        if "logic" in type_lower:
+            channel_names.append(f"D{digital_count}")
+            digital_count += 1
+        elif "v" in type_lower or "dc" in type_lower:
+            # Analog channel
+            channel_names.append(f"A{analog_count}")
+            analog_count += 1
+        else:
+            # Unknown type, use generic name
+            channel_names.append(f"CH{len(channel_names)}")
+
+    return channel_names
+
+
+def _parse_csv_row(channel_names: list[str], row: list[str]) -> dict[str, int | float]:
+    """Parse a CSV data row into channel values.
+
+    Args:
+        channel_names: List of channel names
+        row: List of value strings
+
+    Returns:
+        Dict mapping channel name to parsed value
+    """
+    values = {}
+
+    for channel, value in zip(channel_names, row, strict=True):
+        value = value.strip()
+        # Try to parse as number (analog) or binary (digital)
+        try:
+            if "." in value or "e" in value.lower():
+                values[channel] = float(value)
+            else:
+                values[channel] = int(value)
+        except ValueError:
+            # Keep as string if not a number
+            values[channel] = value
+
+    return values
+

packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/csv_test.py

@@ -0,0 +1,132 @@
+"""Tests for CSV format parser."""
+
+from shutil import which
+
+import pytest
+
+from .client import SigrokClient
+from .common import CaptureConfig, CaptureResult, OutputFormat
+from .driver import Sigrok
+from jumpstarter.common.utils import serve
+
+
+@pytest.fixture
+def demo_driver_instance():
+    """Create a Sigrok driver instance configured for the demo device."""
+    # Demo driver has 8 digital channels (D0-D7) and 5 analog (A0-A4)
+    # Map device channels to decoder-friendly semantic names
+    return Sigrok(
+        driver="demo",
+        executable="sigrok-cli",
+        channels={
+            "D0": "vcc",
+            "D1": "cs",
+            "D2": "miso",
+            "D3": "mosi",
+            "D4": "clk",
+            "D5": "sda",
+            "D6": "scl",
+            "D7": "gnd",
+        },
+    )
+
+
+@pytest.fixture
+def demo_client(demo_driver_instance):
+    """Create a client for the demo Sigrok driver."""
+    with serve(demo_driver_instance) as client:
+        yield client
+
+
+@pytest.mark.skipif(which("sigrok-cli") is None, reason="sigrok-cli not installed")
+def test_csv_format_basic(demo_client: SigrokClient):
+    """Test CSV format capture with demo driver."""
+    cfg = CaptureConfig(
+        sample_rate="50kHz",
+        samples=50,
+        output_format=OutputFormat.CSV,
+        channels=["vcc", "cs"],  # Select specific digital channels
+    )
+
+    result = demo_client.capture(cfg)
+    assert isinstance(result, CaptureResult)
+    assert isinstance(result.data, bytes)
+    decoded_data = result.decode()
+    assert isinstance(decoded_data, list)
+    assert len(decoded_data) > 0
+    # Verify channel names are in the data
+    first_sample = decoded_data[0]
+    assert "D0" in first_sample.values or "D1" in first_sample.values
+
+
+@pytest.mark.skipif(which("sigrok-cli") is None, reason="sigrok-cli not installed")
+def test_csv_format_timing(demo_client: SigrokClient):
+    """Test CSV format timing calculations with integer nanoseconds."""
+    cfg = CaptureConfig(
+        sample_rate="100kHz",
+        samples=50,
+        output_format=OutputFormat.CSV,
+        channels=["D0", "D1", "D2"],  # Select specific channels
+    )
+
+    result = demo_client.capture(cfg)
+    assert isinstance(result, CaptureResult)
+
+    # Decode the CSV data
+    samples = result.decode()
+    assert isinstance(samples, list)
+    assert len(samples) > 0
+
+    # Verify timing progresses correctly
+    for sample in samples:
+        assert isinstance(sample.time_ns, int)
+        # Verify timing progresses (1/100kHz = 10,000ns per sample)
+        assert sample.time_ns == sample.sample * 10_000
+
+
+@pytest.mark.skipif(which("sigrok-cli") is None, reason="sigrok-cli not installed")
+def test_csv_format_analog_channels(demo_client: SigrokClient):
+    """Test CSV capture of analog channels with voltage values."""
+    cfg = CaptureConfig(
+        sample_rate="100kHz",
+        samples=20,
+        output_format=OutputFormat.CSV,
+        channels=["A0", "A1"],  # Select specific analog channels
+    )
+
+    result = demo_client.capture(cfg)
+    assert isinstance(result, CaptureResult)
+    assert isinstance(result.data, bytes)
+    decoded_data = result.decode()
+    assert isinstance(decoded_data, list)
+    assert len(decoded_data) > 0
+
+    # Check first sample for analog values
+    first_sample = decoded_data[0]
+    assert len(first_sample.values) > 0
+
+    # Analog values should be floats (voltages)
+    for _channel, value in first_sample.values.items():
+        assert isinstance(value, (int, float))
+
+
+@pytest.mark.skipif(which("sigrok-cli") is None, reason="sigrok-cli not installed")
+def test_csv_format_mixed_channels(demo_client: SigrokClient):
+    """Test CSV with both digital and analog channels."""
+    cfg = CaptureConfig(
+        sample_rate="100kHz",
+        samples=30,
+        output_format=OutputFormat.CSV,
+        channels=["D0", "D1", "A0"],  # Mix of digital and analog
+    )
+
+    result = demo_client.capture(cfg)
+    samples = result.decode()
+
+    assert isinstance(samples, list)
+    assert len(samples) > 0
+
+    # Verify we have values for channels
+    first_sample = samples[0]
+    assert len(first_sample.values) > 0
+

packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/driver.py

@@ -8,7 +8,7 @@
 from shutil import which
 from tempfile import TemporaryDirectory
 
-from .common import CaptureConfig, DecoderConfig
+from .common import CaptureConfig, DecoderConfig, OutputFormat
 from jumpstarter.driver import Driver, export
 
 
@@ -65,7 +65,7 @@ def get_channel_map(self) -> dict[int, str]:
 
     @export
     def list_output_formats(self) -> list[str]:
-        return ["csv", "srzip", "vcd", "binary", "bits", "ascii"]
+        return OutputFormat.all()
 
     @export
     def capture(self, config: CaptureConfig | dict) -> dict: