sigrok - logic analyzer driver#22
Conversation
📝 WalkthroughWalkthroughThis pull request introduces a new Jumpstarter driver package for Sigrok integration, enabling logic analyzer and protocol capture via sigrok-cli. The package provides device scanning, configurable single-shot and streaming captures with multiple output formats (CSV, VCD, BITS, ASCII), protocol decoder support, and format parsers with automatic timestamp calculations. Changes
Sequence DiagramsequenceDiagram
actor User
participant Client as SigrokClient
participant Driver as Sigrok Driver
participant CLI as sigrok-cli
participant Parser as Format Parser
User->>Client: capture(config)
Client->>Driver: call("capture", config)
Driver->>Driver: _build_capture_command(config)
Note over Driver: Construct CLI args:<br/>channels, sample_rate,<br/>triggers, decoders
Driver->>CLI: Execute with output file
CLI->>CLI: Capture samples
CLI-->>Driver: Write output (CSV/VCD/etc)
Driver->>Driver: Read & base64-encode
Driver-->>Client: CaptureResult (data_b64)
Client-->>User: CaptureResult
User->>Client: result.decode()
Client->>Parser: parse_csv/parse_vcd(data)
Parser->>Parser: Parse format headers
Parser->>Parser: Yield Sample objects
Parser-->>Client: Iterator[Sample]
Client-->>User: Decoded samples
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Actionable comments posted: 7
🤖 Fix all issues with AI agents
In `@python/packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/csv.py`:
- Around line 67-82: The _extract_csv_data_lines function currently returns
blank lines which later produce empty rows and cause zip(..., strict=True) to
raise ValueError; update _extract_csv_data_lines to also skip
empty/whitespace-only lines (after strip) so it only appends non-empty CSV data
lines (i.e., keep the existing comment and analog-preview checks in
_extract_csv_data_lines and add a guard that continues if line == "" to avoid
emitting empty rows).
In
`@python/packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/driver_test.py`:
- Around line 304-305: Replace the strict float equality assertion on
sample.time with an approximate comparison using pytest.approx to avoid
precision errors: change the assertion that compares sample.time to
sample.sample * 0.00001 (the current line referencing sample.time and
sample.sample) to use pytest.approx(...). Ensure pytest is imported in
jumpstarter_driver_sigrok/driver_test.py if not already present and use
pytest.approx(sample.sample * 0.00001) (or an explicit tolerance via rel/abs) in
the assert.
In
`@python/packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/driver.py`:
- Around line 161-182: _channel_args currently returns no -C when self.channels
is empty or when selected_names contain device keys (e.g., "D0") that aren't
values in self.channels, causing selections to be ignored; modify _channel_args
and related logic so selected_names is matched against both user names and
device keys: when self.channels is empty, build channel_map from selected_names
by treating each name as a device (mapping name=name), and when filtering a
non-empty self.channels include entries where either the user name (value) or
the device name (key) is in selected_names; also update _resolve_channel to
accept device names by returning the device key when present (falling back to
existing name->device lookup), ensuring trigger/decoder pin mapping honors
selections and unmapped device names.
- Around line 107-112: The subprocess is created with
stderr=asyncio.subprocess.PIPE but never read, which can deadlock; in the code
around the asyncio.create_subprocess_exec call (search for the
self.logger.debug("streaming sigrok-cli: %s", " ".join(cmd)) and the process =
await asyncio.create_subprocess_exec(...) line in
jumpstarter_driver_sigrok/driver.py), either change stderr to None to inherit
the parent stderr or add a concurrent reader that drains process.stderr (e.g.,
spawn an asyncio.create_task that reads and discards/logs stderr lines) so the
stderr pipe cannot fill and stall the sigrok-cli stream.
In `@python/packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/vcd.py`:
- Around line 22-61: The parser currently only handles value changes when they
appear on the same line as the timestamp; change the main parsing loop (the
second loop that uses sample_idx, timescale_multiplier and channel_map) to treat
a line starting with "#" as a timestamp marker: record the current timestamp,
then iterate subsequent lines until the next "#" (or EOF) and for each
non-empty, non-directive line parse it as a value-change and yield a sample with
sample_idx incremented; additionally handle $dumpvars by detecting a "$dumpvars"
directive and consuming lines until the matching "$end", parsing each inner
value-change the same way. Use/extend the existing helpers (_parse_timescale,
_parse_vcd_timestamp_line or add a new _parse_vcd_value_change) and keep
channel_map/timescale_multiplier usage, assigning sample number via sample_idx
for each yielded change.
In `@python/packages/jumpstarter-driver-sigrok/pyproject.toml`:
- Around line 19-21: The source_archive URL under
[tool.hatch.metadata.hooks.vcs.urls] is pointing to "jumpstarter-dev/repo" which
is incorrect; update the source_archive value to use the correct repository name
"jumpstarter-dev/monorepo" while preserving the {commit_hash} placeholder (i.e.,
change source_archive =
"https://github.com/jumpstarter-dev/repo/archive/{commit_hash}.zip" to reference
"monorepo") so archive links resolve correctly.
In `@python/packages/jumpstarter-driver-sigrok/README.md`:
- Around line 17-28: The README's export YAML for Sigrok is missing the required
config: wrapper; update the example so the sigrok export block nests the driver
settings under config (i.e., export -> sigrok -> type/driver/conn/channels
remain but are placed inside a config: mapping) so it matches the schema used by
examples/exporter.yaml and the Sigrok exporter class
(jumpstarter_driver_sigrok.driver.Sigrok).
❌ Deploy Preview for jumpstarter-docs failed. Why did it fail? →
|
b2d88a4 to
c6cccec
Compare
|
@ambient-code please rebase |
2dab05f to
1e744a3
Compare
|
@kirkbrauer @raballew this is long, but independent of almost everything else. It is extensively tested with the sigrok cli dummy driver. I have also tested it with hardware. It would be a cool addition to 0.9.0 I am happy if the docs and the proposed api makes sense to you. |
|
@ambient-code please handle comments. |
kirkbrauer
left a comment
There was a problem hiding this comment.
I think this looks OK as a first attempt, I asked the @ambient-code to add some additional tests, but I think this is good to merge assuming it has been tested.
raballew
left a comment
There was a problem hiding this comment.
Better, but I need clarification on a few things.
also, VCD tests should not expect channel mapping when not interacting with sigrok-cli, since sigrok-cli is the one performing mappings.
- Skip empty lines in CSV parser to prevent strict-zip failures - Use pytest.approx() for float comparison in tests - Fix stderr pipe deadlock in capture_stream by inheriting stderr - Fix channel selection to handle device names and empty channel maps - Handle multi-line VCD value changes and $dumpvars blocks - Fix source_archive URL to use correct repo name - Add missing config: wrapper in README exporter YAML example Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Extract _parse_vcd_header, _parse_vcd_body, and _parse_timestamp from parse_vcd to bring cyclomatic complexity under the C901 threshold. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
- Add missing driver entry-points registration in pyproject.toml - Add sample_rate field_validator with strict regex to prevent command injection - Convert OutputFormat to str Enum for proper validation - Add DecoderConfig.options validator rejecting ':' in keys/values - Fix _parse_timescale: add 'fs' unit, raise ValueError on unknown units - Remove dead code _parse_vcd_timestamp_line - Fix TemporaryDirectory leak by using context manager in capture() - Add subprocess timeout (configurable, default 300s) to scan() and capture() - Return list instead of generator from decode() for re-iterability - Replace non-ASCII characters (mu -> us, arrow -> ->) - Fix misleading "50 chars" comment to "25 chars" - Log warning when VCD x/z states are mapped to 0 - Remove module-level pytestmark from driver_test.py - Fix tautological assertions in test_vcd_parser_timescale_variations - Use pytest.approx for float equality in csv_test.py Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
- Create conftest.py with shared test fixtures (deduplicated from csv_test.py and driver_test.py) - Create common_test.py with unit tests for Sample._format_time(), CaptureResult.__str__(), CaptureResult.data, CaptureResult._parse_bits(), and OutputFormat.all() - Add CSV parser unit tests in csv_test.py that test parse_csv() directly with inline data (digital-only, analog-only, mixed, empty, sample rates) - Add _ensure_executable error path test in driver_test.py Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
- Fix import ordering in common.py and csv_test.py (ruff I001) - Fix line too long in common.py validator signature (ruff E501) - Remove unused imports in common_test.py (pytest) and csv_test.py (b64encode, Sample) - Simplify output_format.value accesses by removing redundant hasattr checks - Replace non-ASCII arrow (→) with ASCII alternative (->) in driver_test.py Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
106da37 to
ae0195f
Compare
- Fix asyncio.create_subprocess_exec call: use cmd[0], *cmd[1:] so ty can
resolve the required 'program' parameter; use subprocess.PIPE instead of
asyncio.subprocess.PIPE (unresolved attribute per ty)
- Replace bare string literals ("srzip", "csv", "binary") passed as
output_format to CaptureConfig with OutputFormat enum values so ty does
not flag them as invalid-argument-type
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
| if symbol in channel_map: | ||
| try: | ||
| return int(binary_value, 2), symbol, next_pos | ||
| except ValueError: | ||
| return 0, symbol, next_pos |
There was a problem hiding this comment.
[HIGH] The binary value parser maps the entire multi-bit value to 0 when any single bit is x or z. For example, b10x10 hits ValueError on int(binary_value, 2) and the except handler returns 0 for the whole value. A 32-bit bus value with one unknown bit silently becomes zero with no warning.
Suggested fix: replace x/z characters with '0' before calling int(binary_value, 2), and log a warning about the substitution.
AI-generated, human reviewed
| class DecoderConfig(BaseModel): | ||
| """Protocol decoder configuration (real-time during capture).""" | ||
|
|
||
| name: str |
There was a problem hiding this comment.
[HIGH] DecoderConfig.name has no validation. While options rejects colons (lines 81-94), the name is passed directly to the -P flag (segments = [decoder.name] at driver.py:222) and the -A flag (driver.py:234). A name like "uart:rx=D0" would inject into the colon-delimited argument.
Suggested fix: add a field_validator for name with a pattern like ^[a-zA-Z0-9_-]+$ to reject colons, spaces, and special characters.
AI-generated, human reviewed
| def capture(self, config: CaptureConfig | dict) -> CaptureResult: | ||
| return CaptureResult.model_validate(self.call("capture", config)) | ||
|
|
||
| def capture_stream(self, config: CaptureConfig | dict): |
There was a problem hiding this comment.
[HIGH] get_driver_info and get_channel_map use bare dict return types without type parameters, and capture_stream has no return type annotation at all.
Suggested fix: use dict[str, str] for get_channel_map, parameterize the dict for get_driver_info, and annotate capture_stream with -> Generator[bytes, None, None].
AI-generated, human reviewed
| return result.stdout | ||
|
|
||
| @export | ||
| def get_driver_info(self) -> dict: |
There was a problem hiding this comment.
Use a TypedDict or dedicated result type for capture(). Add -> AsyncGenerator[bytes, None] for capture_stream. Parameterize the dict return for get_driver_info.
| 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 |
There was a problem hiding this comment.
Change the annotation to -> dict[str, int | float | str], or remove the string fallback by raising or logging instead.
| # Get channel names from types | ||
| channel_names = _infer_channel_names(types_row) | ||
|
|
||
| # Parse and yield data rows one by one |
| driver.scan() | ||
|
|
||
|
|
||
| @pytest.mark.skipif(which("sigrok-cli") is None, reason="sigrok-cli not installed") |
There was a problem hiding this comment.
The driver passes timeout=self.timeout to subprocess.run(), but no test verifies that timeout enforcement actually works. There is no test that patches subprocess.run to raise TimeoutExpired.
| @field_validator("options") | ||
| @classmethod | ||
| def validate_options(cls, v: dict[str, str | int | float | bool] | None) -> dict[str, str | int | float | bool] | None: # noqa: E501 | ||
| if v is None: | ||
| return v | ||
| for key, value in v.items(): | ||
| str_key = str(key) | ||
| str_value = str(value) | ||
| if ":" in str_key or ":" in str_value: | ||
| raise ValueError( | ||
| f"Decoder option key/value must not contain ':' (got key={str_key!r}, value={str_value!r}). " | ||
| "Colons are used as sigrok-cli delimiters and could cause option injection." | ||
| ) | ||
| return v |
There was a problem hiding this comment.
validate_options (rejects colons for injection prevention) and validate_sample_rate (enforces format) have zero test coverage.
| """Protocol decoder configuration (real-time during capture).""" | ||
|
|
||
| name: str | ||
| channels: dict[str, str] | None = None |
There was a problem hiding this comment.
A channel key containing : would inject into the -P argument. Add a field_validator for channels keys rejecting colons and other delimiter characters.
| @field_validator("sample_rate") | ||
| @classmethod | ||
| def validate_sample_rate(cls, v: str) -> str: | ||
| if not re.match(r"^\d+(\.\d+)?\s*(k|M|G)?(Hz)?$", v): |
There was a problem hiding this comment.
remove \s* from the regex pattern.
|
@mangelajo This PR has been pending for quite a while, I think we should decide if we want to merge it or close it for later. |
Summary by CodeRabbit
Release Notes
New Features
Documentation
Infrastructure
✏️ Tip: You can customize this high-level summary in your review settings.