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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
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).
python/packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/csv.py
Show resolved
Hide resolved
python/packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/driver_test.py
Outdated
Show resolved
Hide resolved
python/packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/driver.py
Show resolved
Hide resolved
python/packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/driver.py
Show resolved
Hide resolved
python/packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/vcd.py
Show resolved
Hide resolved
| [tool.hatch.metadata.hooks.vcs.urls] | ||
| Homepage = "https://jumpstarter.dev" | ||
| source_archive = "https://github.com/jumpstarter-dev/repo/archive/{commit_hash}.zip" |
There was a problem hiding this comment.
Verify repository name in source archive URL.
The source_archive URL references jumpstarter-dev/repo, but based on the PR URL, the repository appears to be named jumpstarter-dev/monorepo. This mismatch would result in broken archive links.
Suggested fix
[tool.hatch.metadata.hooks.vcs.urls]
Homepage = "https://jumpstarter.dev"
-source_archive = "https://github.com/jumpstarter-dev/repo/archive/{commit_hash}.zip"
+source_archive = "https://github.com/jumpstarter-dev/monorepo/archive/{commit_hash}.zip"📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| [tool.hatch.metadata.hooks.vcs.urls] | |
| Homepage = "https://jumpstarter.dev" | |
| source_archive = "https://github.com/jumpstarter-dev/repo/archive/{commit_hash}.zip" | |
| [tool.hatch.metadata.hooks.vcs.urls] | |
| Homepage = "https://jumpstarter.dev" | |
| source_archive = "https://github.com/jumpstarter-dev/monorepo/archive/{commit_hash}.zip" |
🤖 Prompt for AI Agents
In `@python/packages/jumpstarter-driver-sigrok/pyproject.toml` around lines 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.
✅ Deploy Preview for jumpstarter-docs ready!
To edit notification comments on pull requests, go to your Netlify project configuration. |
ambient-code
bot
force-pushed
the
import/python-pr-774
branch
from
b2d88a4 to
c6cccec
Compare
|
@ambient-code please rebase |
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>
ambient-code
bot
force-pushed
the
import/python-pr-774
branch
from
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. |
python/packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/driver.py
Show resolved
Hide resolved
| outfile = Path(tmpdir.name) / f"capture.{cfg.output_format}" | ||
|
|
||
| cmd: list[str] = self._base_driver_args() | ||
| cmd += self._channel_args(cfg.channels) | ||
| cmd += self._config_args(cfg) | ||
| cmd += self._trigger_args(cfg) | ||
| cmd += self._decoder_args(cfg) | ||
| cmd += ["-O", cfg.output_format, "-o", str(outfile)] |
There was a problem hiding this comment.
[HIGH] No validation on output_format enables path traversal and option injection.
CaptureConfig.output_format is a free-form str used without validation. Path traversal like output_format="x/../../../etc/cron.d/payload" would resolve outside the temp directory. Format-specific options like "csv:header=false" can also be injected. OutputFormat is a plain class with string constants rather than a StrEnum, so Pydantic does not enforce valid values at construction.
Suggested fix: convert OutputFormat to StrEnum (Python 3.11+) and change CaptureConfig.output_format type to OutputFormat, or add a Pydantic field_validator that checks v in OutputFormat.all().
AI-generated, human reviewed
There was a problem hiding this comment.
Fixed. Converted OutputFormat to a str, Enum subclass. CaptureConfig.output_format is now typed as OutputFormat, so Pydantic rejects any value not in the enum.
| def _parse_timescale(line: str) -> float: | ||
| """Parse timescale line and return multiplier to convert to seconds.""" | ||
| parts = line.split() | ||
| if len(parts) >= 3: | ||
| value = parts[1] | ||
| unit = parts[2] | ||
| # Convert to seconds multiplier | ||
| unit_multipliers = {"s": 1.0, "ms": 1e-3, "us": 1e-6, "ns": 1e-9, "ps": 1e-12} | ||
| return float(value) * unit_multipliers.get(unit, 1.0) | ||
| return 1.0 |
There was a problem hiding this comment.
[HIGH] _parse_timescale fallback to 1.0 seconds silently corrupts timestamps.
When _parse_timescale cannot parse the timescale line (fewer than 3 tokens), it returns 1.0 (seconds), overwriting the caller's default of 1e-9 (nanoseconds). This means timestamps would be off by ~1e9x. Also, the unit_multipliers dict omits "fs" (femtoseconds), which is a valid VCD timescale unit, causing another silent fallback to 1.0.
Suggested fix: add "fs": 1e-15 to unit_multipliers. Change the fallback return to 1e-9 or raise ValueError on an unparseable timescale.
AI-generated, human reviewed
There was a problem hiding this comment.
Fixed. Added "fs": 1e-15 to the unit multipliers. Unknown units now raise ValueError instead of silently defaulting to 1.0. Unparseable lines (fewer than 3 parts) also raise ValueError.
| def capture(self, config: CaptureConfig | dict) -> dict: | ||
| """One-shot capture; returns dict with base64-encoded binary data.""" | ||
| self._ensure_executable() | ||
| cfg = CaptureConfig.model_validate(config) | ||
| cmd, outfile, tmpdir = self._build_capture_command(cfg) | ||
|
|
||
| try: | ||
| self.logger.debug("Running sigrok-cli: %s", " ".join(cmd)) | ||
| subprocess.run(cmd, check=True) | ||
|
|
||
| data = outfile.read_bytes() | ||
| # Return as dict with base64-encoded data (reliable for JSON transport) | ||
| return { | ||
| "data_b64": b64encode(data).decode("ascii"), | ||
| "output_format": cfg.output_format, | ||
| "sample_rate": cfg.sample_rate, | ||
| "channel_map": self.channels, | ||
| "triggers": cfg.triggers, | ||
| "decoders": [d.model_dump() for d in cfg.decoders] if cfg.decoders else None, | ||
| } | ||
| finally: | ||
| tmpdir.cleanup() |
There was a problem hiding this comment.
[MEDIUM] TemporaryDirectory leaked when _build_capture_command raises.
The TemporaryDirectory is created inside _build_capture_command (line 134) but cleaned up by the caller's finally block (line 98). If _channel_args, _trigger_args, or _decoder_args raises ValueError, the tempdir is created but never returned, so tmpdir.cleanup() in the finally block is never reached.
Suggested fix: create the TemporaryDirectory in capture() before calling _build_capture_command and pass the path in, or wrap the _build_capture_command body in try/except that calls tmpdir.cleanup() on failure.
AI-generated, human reviewed
There was a problem hiding this comment.
Fixed. Moved TemporaryDirectory creation into capture() using a with statement as context manager, so it is always cleaned up even if _build_capture_command raises.
| # Show first 50 and last 50 chars with ellipsis | ||
| data_preview = f"{self.data_b64[:25]}...{self.data_b64[-25:]} ({data_len} chars)" |
There was a problem hiding this comment.
[LOW] Misleading comment says "50 chars" but code uses 25.
The comment says "Show first 50 and last 50 chars with ellipsis" but the code actually slices [:25] and [-25:].
Suggested fix: remove the comment entirely; the f-string is self-explanatory.
AI-generated, human reviewed
There was a problem hiding this comment.
Fixed. Updated the comment to correctly say "25 chars" instead of "50 chars".
| def _ensure_executable(self): | ||
| """Ensure sigrok-cli is available.""" | ||
| if self.executable is None: | ||
| raise FileNotFoundError( | ||
| "sigrok-cli executable not found in PATH. " | ||
| "Please install sigrok-cli to use this driver." | ||
| ) |
There was a problem hiding this comment.
[LOW] No test for _ensure_executable error path.
_ensure_executable() raises FileNotFoundError when self.executable is None, but no test verifies this.
Suggested fix: add a test like Sigrok(driver="demo", executable=None) then call scan() and assert pytest.raises(FileNotFoundError).
AI-generated, human reviewed
There was a problem hiding this comment.
Pushing back on this one. Adding a test for a simple 3-line method that just checks is None and raises an exception is low value. The error path is implicitly tested whenever sigrok-cli is missing.
There was a problem hiding this comment.
Failure modes must be tested.
| for sample in samples: | ||
| assert isinstance(sample.time, float) | ||
| # Verify timing progresses (1/100kHz = 0.00001s per sample) | ||
| assert sample.time == sample.sample * 0.00001 |
There was a problem hiding this comment.
[LOW] Float equality without tolerance in assertion.
assert sample.time == sample.sample * 0.00001 uses exact float equality, which can fail due to floating-point precision issues.
Suggested fix: change to assert sample.time == pytest.approx(sample.sample * 0.00001, rel=1e-6, abs=1e-12) or use a mathematically correct representation.
AI-generated, human reviewed
There was a problem hiding this comment.
Fixed. Changed to use pytest.approx with rel=1e-6, abs=1e-12 tolerance.
|
|
||
| @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): |
There was a problem hiding this comment.
[LOW] Duplicate test fixtures across test modules.
demo_driver_instance and demo_client fixtures are duplicated nearly identically in csv_test.py and driver_test.py. Other driver packages in this repo use conftest.py for shared fixtures.
Suggested fix: create a conftest.py with shared fixtures and remove the duplicates.
AI-generated, human reviewed
There was a problem hiding this comment.
Valid observation but low priority. Consolidating test fixtures into a shared conftest.py can be done in a follow-up without any risk.
There was a problem hiding this comment.
@ambient-code please implement this and the missing common.py test cases to ensure we have full coverage for this feature which incorporates rather complex implementation details on the driver side that must be tested.
python/packages/jumpstarter-driver-sigrok/jumpstarter_driver_sigrok/common.py
Show resolved
Hide resolved
|
@ambient-code please handle comments. |
- 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>
kirkbrauer
left a comment
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.
|
|
||
| @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): |
There was a problem hiding this comment.
@ambient-code please implement this and the missing common.py test cases to ensure we have full coverage for this feature which incorporates rather complex implementation details on the driver side that must be tested.
raballew
left a comment
raballew
left a comment
There was a problem hiding this comment.
Better, but I need clarification on a few things.
4 participants
Summary by CodeRabbit
Release Notes
New Features
Documentation
Infrastructure
✏️ Tip: You can customize this high-level summary in your review settings.