Skip to content

Add dat bypass, build-cluster-db CLI, and incremental improvements - #10

Merged
ypriverol merged 15 commits into
devfrom
feature/dat-bypass-cluster-db
Apr 18, 2026
Merged

ypriverol merged 15 commits into
devfrom
feature/dat-bypass-cluster-db

Conversation

@ypriverol

@ypriverol ypriverol commented Apr 13, 2026 •

Copy link
Copy Markdown
Contributor

Summary

  • Add maracluster_dat.py: parquet-to-dat conversion bypassing MGF entirely (15x smaller output)
  • Add build-cluster-db and parquet2dat CLI commands
  • Expand incremental merge-clusters with BIN strategy support
  • Add QPX metadata module (qpx_metadata.py) and MSNet-to-QPX converter

Details

The dat bypass writes MaRaCluster's internal binary format directly from parquet. Binning replicates BinSpectra::binBinaryTruncated: bin = floor(mz / 1.000508 + 0.32), top-40 peaks, skip peaks >= precursor neutral mass. Size: ~100 bytes/spectrum vs ~1.5 KB in MGF.

The build-cluster-db command creates cluster_metadata.parquet + psm_cluster_membership.parquet from MaRaCluster output + scan_titles mapping, designed for the dat-bypass workflow.

Test plan

  • Run python test_e2e_single_dataset.py on PXD014877
  • Verify pyspectrafuse convert-dat produces valid .dat files
  • Test pyspectrafuse build-cluster-db end-to-end
  • Test incremental merge with --method_type bin

🤖 Generated with Claude Code

Summary by CodeRabbit

Release Notes

  • New Features

    • Added cluster database building with statistical computation and purity analysis.
    • Added QPX format conversion and metadata generation from SDRF files.
    • Added MaRaCluster-compatible binary file conversion for clustering pipelines.
    • Added incremental clustering support for merging new results with existing databases.
    • Updated MSP spectrum export with improved cluster membership handling.
  • Removed Features

    • Removed MGF conversion command; replaced with DAT file output format.
  • Infrastructure

    • Bumped minimum Python requirement to 3.10+; added DuckDB dependency.
    • Updated build tooling and container definitions.

ypriverol and others added 5 commits April 11, 2026 12:01
Normalizes both legacy msnet and QPX parquet formats to canonical column
names at load time, extracts global_qvalue from QPX additional_scores,
and adds a standalone cluster2parquet command for the cluster-parquet step.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…head

BestSpectrumStrategy now uses groupby.idxmin() instead of
groupby.apply(top_n_rows) for selecting the best spectrum per cluster
(~27x measured speedup). Base class uses sort_values + groupby.head()
for top-N selection in large clusters (~76x measured speedup). Also
replaces np.isin with .isin() and np.vstack with pd.concat for
cleaner DataFrame operations.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Previously only the first parquet file was processed when a list was
passed. Now iterates over all files, accumulating cluster results across
the full dataset. State (write counts, file indices) persists across
files so MGF path/order mapping remains consistent.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Implements three-phase incremental clustering for adding new projects
without re-clustering the entire dataset:

1. extract-reps: Writes one representative spectrum per cluster from
   cluster_metadata.parquet to MGF (TITLE=rep:{cluster_id} for tracing)
2. resolve-clusters: After MaRaCluster runs on reps + new data, maps
   new cluster IDs back to existing ones via representative membership.
   Handles cluster merges (multiple old clusters → one new).
3. merge-results: Appends new PSMs to membership table (dedup by USI),
   rebuilds cluster metadata with updated stats and consensus spectra.

CLI: pyspectrafuse incremental extract-reps / merge-clusters
Tests cover extraction, index building, resolution, and merge scenarios.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- Add parquet-to-dat conversion (maracluster_dat.py) bypassing MGF entirely (15x smaller)
- Add build-cluster-db command for dat-bypass workflow
- Add parquet2dat CLI command
- Expand incremental merge-clusters with BIN strategy support
- Update cluster2parquet for QPX metadata compatibility
- Register new CLI commands in pyspectrafuse.py

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
@coderabbitai

coderabbitai Bot commented Apr 13, 2026 •

Copy link
Copy Markdown

Important

Review skipped

Auto reviews are disabled on base/target branches other than the default branch.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 67e510ef-8b5f-4577-a60e-44cef288f765

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

  • 🔍 Trigger review

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

  • @coderabbitai resume to resume automatic reviews.
  • @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

  • ▶️ Resume reviews
  • 🔍 Trigger review
📝 Walkthrough

Walkthrough

This PR transitions the codebase from SDRF-based MGF generation workflows to QPX-based clustering with DuckDB, introduces MaRaCluster binary .dat format support, adds incremental clustering capabilities, and modernizes the build system to hatchling with Python 3.10+ requirements.

Changes

Cohort / File(s) Summary
Removed cluster/parquet handling
pyspectrafuse/cluster_parquet_combine/cluster_res_handler.py, pyspectrafuse/cluster_parquet_combine/combine_cluster_and_parquet.py, pyspectrafuse/mgf_convert/parquet2mgf.py, pyspectrafuse/common/sdrf_utils.py
Deleted legacy cluster-directory traversal, TSV parsing, and MGF generation infrastructure; removed SdrfUtil class for SDRF metadata extraction.
Removed legacy CLI command
pyspectrafuse/commands/quantmsio2mgf.py
Eliminated convert-mgf Click command that performed parquet-to-MGF conversion via multiprocessing.
New cluster database construction
pyspectrafuse/commands/build_cluster_db.py
Added build-cluster-db command implementing DuckDB-backed cluster membership and metadata Parquet generation from MaRaCluster TSV and scan-title files.
New QPX conversion command
pyspectrafuse/commands/convert_msnet_to_qpx.py
Added convert-to-qpx command for MSNet→QPX PSM Parquet conversion and SDRF-derived metadata Parquet generation (.sample.parquet, .run.parquet, .ontology.parquet, .dataset.parquet).
New MaRaCluster binary export
pyspectrafuse/commands/parquet2dat.py
Added convert-dat Click command to export QPX Parquet spectra to MaRaCluster-compatible .dat binary format with optional charge filtering.
New incremental clustering
pyspectrafuse/commands/incremental.py
Added incremental command group with extract-reps-dat subcommand for representative spectrum extraction from cluster metadata.
Updated spectrum-to-MSP
pyspectrafuse/commands/spectrum2msp.py
Refactored to use DuckDB-normalized parquet schema querying, MaRaCluster TSV + scan-title-based cluster mapping, chunked batch processing, and gzip output format; removed SDRF discovery.
New DuckDB utilities
pyspectrafuse/common/duckdb_ops.py
Added comprehensive DuckDB operations: cluster statistics/purity computation, PSM deduplication, spectrum-array scanning, and windowed cluster merging.
New schema definitions
pyspectrafuse/common/schemas.py
Added PyArrow schemas for cluster_metadata.parquet and psm_cluster_membership.parquet outputs.
New QPX metadata reader
pyspectrafuse/common/qpx_metadata.py
Added utilities to parse and merge .run.parquet and .sample.parquet metadata into unified organism/instrument mappings.
Enhanced parquet utilities
pyspectrafuse/common/parquet_utils.py
Updated recursive parquet discovery to filter out QPX metadata files and pipeline output directories.
Updated constant definitions
pyspectrafuse/common/constant.py
Expanded parquet column mappings to support both MSNet and QPX schemas; added ParquetSchemaAdapter for column normalization and scan-type detection.
Updated MSP formatting
pyspectrafuse/common/msp_utils.py
Modified peak-array handling for conditional string evaluation and gzip file writing support.
New MaRaCluster binary module
pyspectrafuse/maracluster_dat.py
Added end-to-end parquet-to-.dat conversion with peak binning, scan info packing, title generation, and m/z-window-based file splitting.
New incremental resolution
pyspectrafuse/incremental/resolve_clusters_dat.py
Added cluster TSV parsing, scan-title aggregation, incremental cluster ID resolution with UUID minting, and spectrum enrichment from parquet PSM data.
New representative extraction
pyspectrafuse/incremental/representative_dat.py
Added extraction of consensus spectra from cluster metadata into MaRaCluster binary format with optional filtering.
New incremental merge
pyspectrafuse/incremental/merge_results.py
Added end-to-end incremental clustering merge: membership append/dedup, consensus recomputation, best-PSM selection, and metadata assembly with provenance tracking.
Updated consensus strategies
pyspectrafuse/consensus_strategy/*.py
Modified groupby behavior across average/best/binning/base/most-similar strategies; replaced include_groups with group_keys, vectorized top-N selection, simplified dataframe concatenation.
Updated CLI registration
pyspectrafuse/pyspectrafuse.py
Removed quantmsio2mgf command; added incremental, convert_to_qpx_cmd, parquet2dat, build_cluster_db_cmd.
Updated build system
pyproject.toml
Switched build backend from setuptools to hatchling; raised Python minimum from 3.8 to 3.10; added DuckDB dependency; updated version to 0.0.4.
Updated Dockerfile
Dockerfile
Replaced pip/requirements.txt with uv pip install; updated metadata version; added CMD entrypoint.
Updated CI/CD workflows
.github/workflows/pyspectrafuse-containers.yml, .github/workflows/tests.yml
Updated trigger branches, action versions, test matrix Python versions (3.8–3.11 → 3.10–3.12), replaced pip with uv; simplified container/module import checks.
Added build scripts
scripts/build_docker.sh, scripts/build_singularity.sh
Added Docker and Singularity/Apptainer build scripts with configurable versioning and multi-platform support.
Updated documentation
README.md
Rewrote feature list and CLI examples to reflect QPX/MaRaCluster/DuckDB workflows; updated Python version requirements; added Docker/Singularity usage; removed conda/CI details.
Updated version constants
pyspectrafuse/__init__.py, pyspectrafuse/consensus_strategy/__init__.py
Bumped __version__ from 0.0.2 to 0.0.4.
Removed/updated tests
tests/test_integration_ftp.py, tests/test_sdrf_utils.py, tests/test_cli.py, tests/test_incremental.py, tests/conftest.py
Removed legacy SDRF and FTP integration tests; added incremental clustering test coverage; updated CLI test expectations; adjusted SDRF fixture column naming.

Sequence Diagram(s)

sequenceDiagram
    participant User
    participant build_cluster_db as build-cluster-db Command
    participant Parser as TSV/ScanTitle Parser
    participant DuckDB
    participant Membership as Membership Parquet
    participant Metadata as Metadata Parquet

    User->>build_cluster_db: provide cluster TSV, scan_titles[], parquet[]
    build_cluster_db->>Parser: parse_cluster_tsv, parse_scan_titles
    Parser-->>build_cluster_db: cluster rows, spectrum mapping
    
    build_cluster_db->>DuckDB: union PSM parquets, filter by charge
    build_cluster_db->>DuckDB: LEFT JOIN spectra to cluster membership
    build_cluster_db->>DuckDB: compute_stats_and_purity per cluster
    DuckDB-->>Membership: write psm_cluster_membership.parquet
    
    build_cluster_db->>DuckDB: select best PSM per cluster
    build_cluster_db->>build_cluster_db: load consensus arrays in chunks
    build_cluster_db->>build_cluster_db: derive peptide_sequence from peptidoform
    build_cluster_db->>Metadata: stream rows to cluster_metadata.parquet
    
    Metadata-->>User: cluster_metadata.parquet + psm_cluster_membership.parquet
Loading
sequenceDiagram
    participant User
    participant parquet2dat as convert-dat Command
    participant ParquetScanner
    participant DatConverter as parquet_to_dat()
    participant BinningEngine
    participant DatWriter as Binary Writer

    User->>parquet2dat: provide parquet_dir, output_dir, [charge_filter]
    parquet2dat->>ParquetScanner: find_target_ext_files(parquet_dir, .parquet)
    ParquetScanner-->>parquet2dat: parquet file list
    
    loop per parquet file
        parquet2dat->>DatConverter: call parquet_to_dat with batch_size
        DatConverter->>DatConverter: stream parquet in batches
        DatConverter->>BinningEngine: compute peak bins per spectrum
        BinningEngine-->>DatConverter: bin indices
        DatConverter->>DatWriter: pack Spectrum/ScanInfo structs
        DatWriter-->>DatConverter: write .dat, .scan_info.dat, .scan_titles.txt
    end
    
    DatWriter-->>User: aggregated counts + output directory
Loading
sequenceDiagram
    participant User
    participant extract_cmd as extract-reps-dat Command
    participant Reader as Parquet Reader
    participant Validator
    participant BinningEngine
    participant DatWriter

    User->>extract_cmd: provide cluster_metadata path, output_dir, [charge_filter, max_clusters]
    extract_cmd->>Reader: batch-iterate cluster_metadata.parquet
    
    loop per batch
        Reader-->>Validator: cluster rows (precursor_mz, charge, consensus arrays)
        Validator->>Validator: validate charge > 0, arrays non-empty
        Validator->>BinningEngine: convert arrays to float64, compute peak bins
        BinningEngine-->>DatWriter: bin indices + pack struct
        DatWriter->>DatWriter: write .dat, .scan_info.dat, .scan_titles.txt (rep:<cluster_id>)
    end
    
    DatWriter-->>User: (written, skipped, dat_path, scaninfo_path, titles_path)
Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related issues

  • SDRF/techsdrf#2: The PR removes all SDRF utilities and SDRF-based workflows, replacing them with QPX parquet metadata (.run.parquet, .sample.parquet) and DuckDB-driven operations, directly addressing legacy SDRF infrastructure concerns.

Possibly related PRs

  • major performance improvements #4: Both modify cluster_parquet_combine and MGF-handling code; this PR deletes ClusterResHandler and related classes that PR #4 had previously optimized.
  • the latest version #2: The PR removes ClusterResHandler, CombineCluster2Parquet, and Parquet2Mgf classes from the same files where PR #2 adds/modifies those exact entities.
  • major changes in the structure of the package #3: Overlapping modifications to cluster_parquet_combine, consensus strategies, SDRF utilities, and CI/packaging configuration.

Suggested labels

enhancement, breaking-change, refactoring, cli, duckdb, maracluster

Poem

🐰 A hop through the QPX garden,
Where clusters bloom with DuckDB grace,
Old SDRF paths fade to starlight,
As .dat files dance through MaRa's embrace—
From binary peaks to consensus dreams,
Our spectrum-sorting rabbit bounds ahead! ✨

🚥 Pre-merge checks | ✅ 3
✅ Passed checks (3 passed)
Check name Status Explanation
Description Check ✅ Passed Check skipped - CodeRabbit’s high-level summary is enabled.
Title check ✅ Passed The title 'Add dat bypass, build-cluster-db CLI, and incremental improvements' clearly and concisely summarizes the main changes: introduction of a DAT format bypass, a new build-cluster-db command, and incremental workflow enhancements.
Docstring Coverage ✅ Passed Docstring coverage is 89.76% which is sufficient. The required threshold is 80.00%.

✏️ Tip: You can configure your own custom pre-merge checks in the settings.

✨ Finishing Touches
🧪 Generate unit tests (beta)
  • Create PR with unit tests
  • Commit unit tests in branch feature/dat-bypass-cluster-db

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.

❤️ Share

Comment @coderabbitai help to get the list of available commands and usage tips.

- Lead with convert-dat and build-cluster-db as recommended commands
- Add incremental clustering and QPX metadata to feature list
- Update project structure with new modules
- Mark MGF conversion and SDRF handling as legacy
- Update version to 0.0.4

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actionable comments posted: 7

🧹 Nitpick comments (2)
pyspectrafuse/commands/parquet2dat.py (1)

35-39: Move parquet discovery out of the spectrum2msp command module.

This command now depends on another CLI command for a generic filesystem helper. That makes command imports bidirectional over time and duplicates logic that already belongs in pyspectrafuse.common.parquet_utils. Pull the helper into common and reuse it from both commands.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/parquet2dat.py` around lines 35 - 39, The parquet
discovery helper find_target_ext_files is being imported from
pyspectrafuse.commands.spectrum2msp causing a cross-command dependency; move or
reuse the generic helper from pyspectrafuse.common.parquet_utils instead: remove
the import of find_target_ext_files from pyspectrafuse.commands.spectrum2msp in
parquet2dat.py and import the equivalent function (or add a new helper) from
pyspectrafuse.common.parquet_utils, then call that helper to produce
parquet_files and delete any duplicated discovery logic so both spectrum2msp and
parquet2dat rely on the single common helper (update exports in parquet_utils if
needed).
pyspectrafuse/commands/spectrum2msp.py (1)

40-53: Return parquet files in a stable order.

glob.glob() does not guarantee a reproducible traversal order. Sorting here keeps downstream behavior deterministic — convert-dat increments file_idx per discovered file, and this command also uses the first entry for output naming.

Proposed fix
-    for path_str in glob.glob(str(Path(directory) / f'**/*{extensions}'), recursive=True):
+    for path_str in sorted(glob.glob(str(Path(directory) / f'**/*{extensions}'), recursive=True)):
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/spectrum2msp.py` around lines 40 - 53, glob.glob() can
return files in nondeterministic order; make the discovery deterministic by
sorting the found paths. Change the loop over glob.glob(...) to iterate over
sorted(glob.glob(...)) or sort the collected list `res` (e.g., call `res.sort()`
before returning) so that the order of `path_str` items and the resulting `res`
list is stable for downstream `file_idx` increments and output naming; update
code around the for-loop that builds `res` (referencing `path_str`, `file`, and
`res`) to ensure the sorted behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@pyspectrafuse/commands/build_cluster_db.py`:
- Around line 272-275: The groupby().idxmin() call can fail when a cluster's
posterior_error_probability is all NaN; before computing best_idx, coerce
posterior_error_probability to numeric on the enriched DataFrame and replace
NaNs with np.inf (e.g., set enriched['_pep'] =
pd.to_numeric(enriched['posterior_error_probability'],
errors='coerce').fillna(np.inf) or call enriched['_pep'].fillna(np.inf,
inplace=True)) so that enriched.groupby('cluster_id')['_pep'].idxmin() returns
valid positions; then compute best_idx and select best_rows as before (symbols:
enriched, '_pep', posterior_error_probability, idxmin, best_idx, best_rows).

In `@pyspectrafuse/commands/quantmsio2mgf.py`:
- Around line 46-53: The worker pool is being created with
Pool(processes=os.cpu_count()) which ignores the --task_parallel option; change
the Pool creation to use the task_parallel value (i.e.
Pool(processes=task_parallel)) so that Parquet2Mgf.convert_to_mgf_task actually
runs with the requested number of worker processes, and keep the existing
chunksize parameter for batching (or compute a separate chunksize variable) when
calling pool.imap to control task batching rather than concurrency.

In `@pyspectrafuse/commands/spectrum2msp.py`:
- Line 198: The code currently calls ParquetPathHandler(...).get_item_info()
with the first parquet file path (path_parquet_lst[0]), which extracts the
accession from the last path segment and therefore yields the parquet filename
instead of the project basename; change the call to pass the project directory
(e.g., the parent directory of the parquet files or the existing project_dir
variable) into ParquetPathHandler so get_item_info() reads the directory name as
the accession; update the reference at the call site (where basename is set) to
construct the handler with the folder containing the parquet files rather than
the file path itself.

In `@pyspectrafuse/common/parquet_utils.py`:
- Around line 58-68: The loop building parquet_path_lst currently appends any
glob hit (parquet_file) without checking whether it's a file, so partitioned
datasets stored as directories ending in .parquet get returned; update the loop
in this helper to skip directories by calling parquet_file.is_file() (or
equivalently ensure pathlib.Path(parquet_file).is_file()) before applying the
existing suffix/name/parts filters and appending to parquet_path_lst, leaving
all other filtering (checks against _metadata_suffixes, _exclude_names,
_exclude_dirs) unchanged.

In `@pyspectrafuse/incremental/resolve_clusters.py`:
- Around line 99-103: The current filtering uses rep_mgf_basename and
str.contains, which yields false positives; instead compute exact basenames
using pathlib.Path and do an equality match: derive rep_mgf_basename from
Path(rep_mgf_path).name, compare against cluster_df['mgf_path'].map(lambda p:
Path(p).name) (or add a temporary column of basenames) to build rep_mask, then
set rep_rows/new_rows accordingly; ensure pathlib.Path is imported if not
already.

In `@pyspectrafuse/mgf_convert/parquet2mgf.py`:
- Around line 155-163: _get_mgf_path currently always pulls instrument from
sample_info_dict, ignoring the skip_instrument option; update _get_mgf_path to
check the skip_instrument flag (the variable/attribute used by Parquet2Mgf for
that option) and, when True, use a single shared instrument bucket (e.g., a
fixed name like "all_instruments" or an empty instrument segment) instead of
info[0]; modify the function where mgf_group_df['mgf_file_path'] is set so
_get_mgf_path reads skip_instrument and constructs the path accordingly
(reference function name _get_mgf_path, the sample_info_dict lookup, and
Parquet2Mgf.skip_instrument or the local skip_instrument variable).

In `@tests/test_incremental.py`:
- Around line 150-155: The fixture string assignments for the variable
tsv_content use unnecessary f-string prefixes (e.g., f"reps.mgf\t0\tNEW_A\n"
blocks) which trigger Ruff F541; remove the leading f from each quoted literal
(including the second occurrence around lines 193-196) so they become plain
string literals while preserving the content and escape sequences.

---

Nitpick comments:
In `@pyspectrafuse/commands/parquet2dat.py`:
- Around line 35-39: The parquet discovery helper find_target_ext_files is being
imported from pyspectrafuse.commands.spectrum2msp causing a cross-command
dependency; move or reuse the generic helper from
pyspectrafuse.common.parquet_utils instead: remove the import of
find_target_ext_files from pyspectrafuse.commands.spectrum2msp in parquet2dat.py
and import the equivalent function (or add a new helper) from
pyspectrafuse.common.parquet_utils, then call that helper to produce
parquet_files and delete any duplicated discovery logic so both spectrum2msp and
parquet2dat rely on the single common helper (update exports in parquet_utils if
needed).

In `@pyspectrafuse/commands/spectrum2msp.py`:
- Around line 40-53: glob.glob() can return files in nondeterministic order;
make the discovery deterministic by sorting the found paths. Change the loop
over glob.glob(...) to iterate over sorted(glob.glob(...)) or sort the collected
list `res` (e.g., call `res.sort()` before returning) so that the order of
`path_str` items and the resulting `res` list is stable for downstream
`file_idx` increments and output naming; update code around the for-loop that
builds `res` (referencing `path_str`, `file`, and `res`) to ensure the sorted
behavior.
🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

  • Push a commit to this branch (recommended)
  • Create a new PR with the fixes

ℹ️ Review info
⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 2b66a7c3-de12-4302-8ff6-4d095e94000f

📥 Commits

Reviewing files that changed from the base of the PR and between 891c6d9 and 3116dcc.

📒 Files selected for processing (27)
  • pyspectrafuse/cluster_parquet_combine/cluster_res_handler.py
  • pyspectrafuse/cluster_parquet_combine/combine_cluster_and_parquet.py
  • pyspectrafuse/commands/build_cluster_db.py
  • pyspectrafuse/commands/cluster2parquet.py
  • pyspectrafuse/commands/convert_msnet_to_qpx.py
  • pyspectrafuse/commands/incremental.py
  • pyspectrafuse/commands/parquet2dat.py
  • pyspectrafuse/commands/quantmsio2mgf.py
  • pyspectrafuse/commands/spectrum2msp.py
  • pyspectrafuse/common/constant.py
  • pyspectrafuse/common/msp_utils.py
  • pyspectrafuse/common/parquet_utils.py
  • pyspectrafuse/common/qpx_metadata.py
  • pyspectrafuse/common/sdrf_utils.py
  • pyspectrafuse/consensus_strategy/average_spectrum_strategy.py
  • pyspectrafuse/consensus_strategy/best_spetrum_strategy.py
  • pyspectrafuse/consensus_strategy/binning_strategy.py
  • pyspectrafuse/consensus_strategy/consensus_strategy_base.py
  • pyspectrafuse/consensus_strategy/most_similar_strategy.py
  • pyspectrafuse/incremental/__init__.py
  • pyspectrafuse/incremental/merge_results.py
  • pyspectrafuse/incremental/representative_mgf.py
  • pyspectrafuse/incremental/resolve_clusters.py
  • pyspectrafuse/maracluster_dat.py
  • pyspectrafuse/mgf_convert/parquet2mgf.py
  • pyspectrafuse/pyspectrafuse.py
  • tests/test_incremental.py
👮 Files not reviewed due to content moderation or server errors (10)
  • pyspectrafuse/common/msp_utils.py
  • pyspectrafuse/common/constant.py
  • pyspectrafuse/common/qpx_metadata.py
  • pyspectrafuse/consensus_strategy/binning_strategy.py
  • pyspectrafuse/pyspectrafuse.py
  • pyspectrafuse/cluster_parquet_combine/cluster_res_handler.py
  • pyspectrafuse/consensus_strategy/average_spectrum_strategy.py
  • pyspectrafuse/common/sdrf_utils.py
  • pyspectrafuse/commands/convert_msnet_to_qpx.py
  • pyspectrafuse/commands/incremental.py

Comment thread pyspectrafuse/commands/quantmsio2mgf.py Resolved
Comment thread pyspectrafuse/commands/spectrum2msp.py Resolved
Comment on lines +58 to 68
# Use glob.glob (follows symlinks) instead of Path.rglob (does not)
for path_str in glob.glob(str(Path(dir_path) / '**/*.parquet'), recursive=True):
parquet_file = Path(path_str)
name = str(parquet_file)
if any(name.endswith(s) for s in _metadata_suffixes):
continue
if any(ex in parquet_file.stem for ex in _exclude_names):
continue
if any(part in _exclude_dirs for part in parquet_file.parts):
continue
parquet_path_lst.append(parquet_file)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Filter out .parquet directories before returning results.

Line 60 turns every glob hit into a candidate path, but this helper never checks is_file(). Partitioned parquet datasets stored as directories named *.parquet will now be returned here and later consumers will treat them as files.

Proposed fix
         # Use glob.glob (follows symlinks) instead of Path.rglob (does not)
         for path_str in glob.glob(str(Path(dir_path) / '**/*.parquet'), recursive=True):
             parquet_file = Path(path_str)
+            if not parquet_file.is_file():
+                continue
             name = str(parquet_file)
             if any(name.endswith(s) for s in _metadata_suffixes):
                 continue
📝 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.

Suggested change
# Use glob.glob (follows symlinks) instead of Path.rglob (does not)
for path_str in glob.glob(str(Path(dir_path) / '**/*.parquet'), recursive=True):
parquet_file = Path(path_str)
name = str(parquet_file)
if any(name.endswith(s) for s in _metadata_suffixes):
continue
if any(ex in parquet_file.stem for ex in _exclude_names):
continue
if any(part in _exclude_dirs for part in parquet_file.parts):
continue
parquet_path_lst.append(parquet_file)
# Use glob.glob (follows symlinks) instead of Path.rglob (does not)
for path_str in glob.glob(str(Path(dir_path) / '**/*.parquet'), recursive=True):
parquet_file = Path(path_str)
if not parquet_file.is_file():
continue
name = str(parquet_file)
if any(name.endswith(s) for s in _metadata_suffixes):
continue
if any(ex in parquet_file.stem for ex in _exclude_names):
continue
if any(part in _exclude_dirs for part in parquet_file.parts):
continue
parquet_path_lst.append(parquet_file)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/common/parquet_utils.py` around lines 58 - 68, The loop
building parquet_path_lst currently appends any glob hit (parquet_file) without
checking whether it's a file, so partitioned datasets stored as directories
ending in .parquet get returned; update the loop in this helper to skip
directories by calling parquet_file.is_file() (or equivalently ensure
pathlib.Path(parquet_file).is_file()) before applying the existing
suffix/name/parts filters and appending to parquet_path_lst, leaving all other
filtering (checks against _metadata_suffixes, _exclude_names, _exclude_dirs)
unchanged.

Comment thread tests/test_incremental.py Resolved
ypriverol and others added 3 commits April 13, 2026 20:59
Parquet-to-dat is the standard conversion path, not a bypass of anything.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- Fix test_get_metadata_dict_from_sdrf: column name case mismatch
  (Characteristics[organism] -> characteristics[organism])
- Switch from setuptools to hatchling build backend
- Bump version to 0.0.4, require Python >=3.10
- Update Dockerfile to use uv for faster installs

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- Replace per-row iterrows() with columnar array access
- Buffer dat/scaninfo/title writes per batch instead of per spectrum
- Pre-extract columns (charges, mz, runs, etc.) outside inner loop
- Same output, 3-5x faster on large datasets

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actionable comments posted: 6

🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@Dockerfile`:
- Around line 39-44: The Docker build fails because pyproject.toml declares
readme = "README.md" but README.md is not copied before running RUN uv pip
install --system --no-cache . Update the Dockerfile so that README.md is copied
into the image before the install step (i.e., place a COPY README.md . between
the COPY pyproject.toml . and COPY pyspectrafuse/ ./ lines) so the package
installation can read the declared README during build.
- Line 34: Replace the non-reproducible image reference used as a build stage
source in the COPY statement (currently "ghcr.io/astral-sh/uv:latest" in the
COPY --from=... line) with an explicit version tag and its SHA256 digest (e.g.,
ghcr.io/astral-sh/uv:<version>@sha256:<digest>); obtain the correct tag and
digest using docker manifest inspect (or the uv Docker integration docs) and
update the COPY --from=... source to that pinned reference to ensure
reproducible, verifiable builds.

In `@pyspectrafuse/maracluster_dat.py`:
- Around line 147-149: The code defines a columns list including 'rt' but still
writes Spectrum records with retentionTime=0.0; update the Spectrum/dict
construction to use the parsed rt value instead of the hard-coded 0.0 (e.g.,
retentionTime=rt or retentionTime=row['rt']), and make the same change at the
other spot flagged (the second Spectrum/dict creation). Ensure the 'rt' value is
cast to float if necessary before assigning to retentionTime so the parquet
persists the actual retention time.
- Around line 196-218: The code converts mz_array and intensity_array to numpy
arrays then calls _bin_spectrum but doesn't verify they have the same length;
when they differ _bin_spectrum may index out-of-bounds. Before computing
charge/prec_mz/etc., add a validation step comparing lengths of mz_arr and
int_arr (and that both lengths > 0) and skip the spectrum (increment skipped) if
they differ or are zero-length; reference the variables mz_arr, int_arr and the
called function _bin_spectrum when adding this guard so only well-formed arrays
reach _bin_spectrum.
- Around line 264-279: The converter is currently grabbing any ".parquet" files
(via find_target_ext_files) and produces .dat outputs for non-spectrum files;
restrict it to spectrum inputs by changing the finder/filter to only match
".psm.parquet" (e.g., call find_target_ext_files(parquet_dir, '.psm.parquet') or
post-filter parquet_files to keep only names ending with '.psm.parquet'), and
keep the FileNotFoundError when the filtered list is empty; reference the
existing parquet_files variable and the find_target_ext_files call so the filter
change is localized and prevents producing .dat for cluster metadata or
membership files.

In `@README.md`:
- Line 162: The fenced code block at the README's project tree is missing a
language identifier (triggering markdownlint MD040); update the opening fence
from ``` to ```text (or another appropriate language like ```bash) so the block
becomes labeled (e.g., change the block around the pyspectrafuse tree to start
with ```text) and leave the closing ``` as-is to satisfy the linter and improve
rendering.
🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

  • Push a commit to this branch (recommended)
  • Create a new PR with the fixes

ℹ️ Review info
⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 9eb047c6-ab59-4739-9146-a66082074b31

📥 Commits

Reviewing files that changed from the base of the PR and between 3116dcc and a2fa10a.

📒 Files selected for processing (5)
  • Dockerfile
  • README.md
  • pyproject.toml
  • pyspectrafuse/maracluster_dat.py
  • tests/conftest.py

Comment thread Dockerfile
&& rm -rf /var/lib/apt/lists/*

## Install uv
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🌐 Web query:

What is the recommended secure Docker pattern for ghcr.io/astral-sh/uv (pinning by explicit version and sha256 digest)?

💡 Result:

The recommended secure Docker pattern for ghcr.io/astral-sh/uv is to pin to an explicit version tag and further pin by SHA256 digest for reproducibility and security against tag movement. Official uv documentation provides this example in a multi-stage Dockerfile: FROM python:3.12-slim-trixie COPY --from=ghcr.io/astral-sh/uv:0.11.6 /uv /uvx /bin/ For maximum security, replace the tag with @sha256:: COPY --from=ghcr.io/astral-sh/uv@sha256:2381d6aa60c326b71fd40023f921a0a3b8f91b14d5db6b90402e65a635053709 /uv /uvx /bin/ To obtain the current digest for a specific tag (e.g., :0.4.17), run: docker manifest inspect ghcr.io/astral-sh/uv:0.4.17 (look for "digest" under the platform config). Available tags include :latest, :{version} (e.g., 0.11.6), and variants like :debian, :python3.12-trixie. View all at https://ghcr.io/astral-sh/uv. Images are signed with GitHub attestations; verify with gh attestation verify oci://ghcr.io/astral-sh/uv:0.11.6. This ensures supply chain security by preventing image substitution.

Citations:


Pin the uv image by explicit version tag and SHA256 digest.

The ghcr.io/astral-sh/uv:latest tag is non-reproducible and weakens supply-chain control. Use an explicit version tag combined with SHA256 digest pinning to ensure build reproducibility and prevent image substitution attacks.

Proposed change
-COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
+COPY --from=ghcr.io/astral-sh/uv:<pinned-version>@sha256:<pinned-digest> /uv /uvx /bin/

Consult the official uv Docker integration guide for available version tags and instructions to retrieve the corresponding digest using docker manifest inspect.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@Dockerfile` at line 34, Replace the non-reproducible image reference used as
a build stage source in the COPY statement (currently
"ghcr.io/astral-sh/uv:latest" in the COPY --from=... line) with an explicit
version tag and its SHA256 digest (e.g.,
ghcr.io/astral-sh/uv:<version>@sha256:<digest>); obtain the correct tag and
digest using docker manifest inspect (or the uv Docker integration docs) and
update the COPY --from=... source to that pinned reference to ensure
reproducible, verifiable builds.

Comment thread Dockerfile Resolved
Comment on lines +147 to +149
columns = ['mz_array', 'intensity_array', 'charge',
'observed_mz', 'calculated_mz', 'rt', 'scan',
'run_file_name', 'peptidoform']

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Persist the parquet RT instead of hard-coding 0.0.

rt is loaded here, but every Spectrum record is still written with retentionTime=0. Any MaRaCluster mode that consumes RT from the .dat payload will see all spectra at time zero.

Proposed fix
             runs = df['run_file_name'].values if 'run_file_name' in df.columns else [''] * len(df)
             peps = df['peptidoform'].values if 'peptidoform' in df.columns else [''] * len(df)
             scans = df['scan'].values if 'scan' in df.columns else [0] * len(df)
+            rts = df['rt'].values if 'rt' in df.columns else np.zeros(len(df), dtype=np.float32)
...
+                rt = rts[i]
+                rt = 0.0 if rt is None or np.isnan(rt) else float(rt)
                 dat_buf.extend(_pack_spectrum(
-                    file_idx, scannr, charge, prec_mz, 0.0, peak_bins))
+                    file_idx, scannr, charge, prec_mz, rt, peak_bins))

Also applies to: 226-227

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/maracluster_dat.py` around lines 147 - 149, The code defines a
columns list including 'rt' but still writes Spectrum records with
retentionTime=0.0; update the Spectrum/dict construction to use the parsed rt
value instead of the hard-coded 0.0 (e.g., retentionTime=rt or
retentionTime=row['rt']), and make the same change at the other spot flagged
(the second Spectrum/dict creation). Ensure the 'rt' value is cast to float if
necessary before assigning to retentionTime so the parquet persists the actual
retention time.

Comment on lines +196 to +218
mz_arr = np.asarray(mz_arr, dtype=np.float64)
int_arr = np.asarray(int_arr, dtype=np.float64)

if len(mz_arr) == 0 or len(int_arr) == 0:
skipped += 1
continue

charge = int(charges[i])
if charge <= 0:
skipped += 1
continue

# Precursor m/z: prefer observed_mz, fall back to calculated_mz
prec_mz = obs_mz[i] if obs_mz is not None else None
if prec_mz is None or np.isnan(prec_mz):
prec_mz = calc_mz[i] if calc_mz is not None else 0.0
prec_mz = float(prec_mz)

# Neutral mass for peak filtering
prec_mass = prec_mz * charge - PROTON_MASS * (charge - 1)

# Bin the spectrum
peak_bins = _bin_spectrum(mz_arr, int_arr, prec_mass)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Validate fragment array lengths before binning.

If a parquet row has different-length mz_array and intensity_array, _bin_spectrum() can index past the end of mz_array and kill the whole conversion instead of skipping one bad spectrum.

Proposed fix
-                if len(mz_arr) == 0 or len(int_arr) == 0:
+                if len(mz_arr) == 0 or len(int_arr) == 0 or len(mz_arr) != len(int_arr):
                     skipped += 1
                     continue
📝 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.

Suggested change
mz_arr = np.asarray(mz_arr, dtype=np.float64)
int_arr = np.asarray(int_arr, dtype=np.float64)
if len(mz_arr) == 0 or len(int_arr) == 0:
skipped += 1
continue
charge = int(charges[i])
if charge <= 0:
skipped += 1
continue
# Precursor m/z: prefer observed_mz, fall back to calculated_mz
prec_mz = obs_mz[i] if obs_mz is not None else None
if prec_mz is None or np.isnan(prec_mz):
prec_mz = calc_mz[i] if calc_mz is not None else 0.0
prec_mz = float(prec_mz)
# Neutral mass for peak filtering
prec_mass = prec_mz * charge - PROTON_MASS * (charge - 1)
# Bin the spectrum
peak_bins = _bin_spectrum(mz_arr, int_arr, prec_mass)
mz_arr = np.asarray(mz_arr, dtype=np.float64)
int_arr = np.asarray(int_arr, dtype=np.float64)
if len(mz_arr) == 0 or len(int_arr) == 0 or len(mz_arr) != len(int_arr):
skipped += 1
continue
charge = int(charges[i])
if charge <= 0:
skipped += 1
continue
# Precursor m/z: prefer observed_mz, fall back to calculated_mz
prec_mz = obs_mz[i] if obs_mz is not None else None
if prec_mz is None or np.isnan(prec_mz):
prec_mz = calc_mz[i] if calc_mz is not None else 0.0
prec_mz = float(prec_mz)
# Neutral mass for peak filtering
prec_mass = prec_mz * charge - PROTON_MASS * (charge - 1)
# Bin the spectrum
peak_bins = _bin_spectrum(mz_arr, int_arr, prec_mass)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/maracluster_dat.py` around lines 196 - 218, The code converts
mz_array and intensity_array to numpy arrays then calls _bin_spectrum but
doesn't verify they have the same length; when they differ _bin_spectrum may
index out-of-bounds. Before computing charge/prec_mz/etc., add a validation step
comparing lengths of mz_arr and int_arr (and that both lengths > 0) and skip the
spectrum (increment skipped) if they differ or are zero-length; reference the
variables mz_arr, int_arr and the called function _bin_spectrum when adding this
guard so only well-formed arrays reach _bin_spectrum.

Comment on lines +264 to +279
"""Convert all .psm.parquet files in a directory to .dat format.

Args:
parquet_dir: Directory containing .psm.parquet files
output_dir: Directory to write .dat files
file_idx_start: Starting file index
charge_filter: If set, only include spectra with this charge

Returns:
Tuple of (list of .dat file paths, total spectra written)
"""
from pyspectrafuse.commands.spectrum2msp import find_target_ext_files

parquet_files = find_target_ext_files(parquet_dir, '.parquet')
if not parquet_files:
raise FileNotFoundError(f"No parquet files found in {parquet_dir}")

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Restrict dataset conversion to the spectrum parquet inputs.

The docstring says this path converts .psm.parquet files, but find_target_ext_files(parquet_dir, '.parquet') will also pick up outputs like cluster_metadata.parquet and psm_cluster_membership.parquet. Those files do not have the spectrum arrays this converter expects, so this silently generates useless .dat artifacts for non-spectrum inputs.

Proposed fix
-    parquet_files = find_target_ext_files(parquet_dir, '.parquet')
+    parquet_files = find_target_ext_files(parquet_dir, '.psm.parquet')
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/maracluster_dat.py` around lines 264 - 279, The converter is
currently grabbing any ".parquet" files (via find_target_ext_files) and produces
.dat outputs for non-spectrum files; restrict it to spectrum inputs by changing
the finder/filter to only match ".psm.parquet" (e.g., call
find_target_ext_files(parquet_dir, '.psm.parquet') or post-filter parquet_files
to keep only names ending with '.psm.parquet'), and keep the FileNotFoundError
when the filtered list is empty; reference the existing parquet_files variable
and the find_target_ext_files call so the filter change is localized and
prevents producing .dat for cluster metadata or membership files.

Comment thread README.md Resolved
ypriverol and others added 2 commits April 17, 2026 17:26
…ental dat modules

- common/duckdb_ops.py + common/schemas.py: centralize cluster DB aggregation
  and parquet schemas; all 11 duckdb ops use try/finally for connection safety
- commands/spectrum2msp.py: rewrite to scan_titles mapping (dat-bypass path);
  drops legacy MGF-based loader and CombineCluster2Parquet coupling
- commands/build_cluster_db.py: DuckDB purity + chunked cluster loading to
  avoid OOM on 3M+ PSM charges
- incremental/representative_dat.py + resolve_clusters_dat.py: dat-based
  representative extraction and cluster ID resolution
- maracluster_dat.py: vectorized batch I/O, window-overlap guard
- Delete MGF path: quantmsio2mgf, parquet2mgf, representative_mgf,
  resolve_clusters (legacy)
- Tests: convert-dat CLI tests, incremental subcommand help tests
- Bump to 0.0.4; pin requirements; Dockerfile CMD (no ENTRYPOINT) for Nextflow

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
- tests.yml: update test matrix for convert-dat CLI
- pyspectrafuse-containers.yml: push 0.0.4 tag to ghcr.io

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actionable comments posted: 10

♻️ Duplicate comments (3)
pyspectrafuse/maracluster_dat.py (3)

290-292: ⚠️ Potential issue | 🟠 Major

find_target_ext_files(..., '.parquet') still pulls non-spectrum parquet files — previous comment not addressed.

This path still accepts any .parquet in the directory, including cluster_metadata.parquet and psm_cluster_membership.parquet. Those won't have mz_array/intensity_array, so parquet_to_dat will silently emit empty/zero-spectrum .dat artifacts.

Proposed fix
-    parquet_files = find_target_ext_files(parquet_dir, '.parquet')
+    parquet_files = find_target_ext_files(parquet_dir, '.psm.parquet')
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/maracluster_dat.py` around lines 290 - 292, The current use of
find_target_ext_files(parquet_dir, '.parquet') collects non-spectrum parquet
files (e.g., cluster_metadata.parquet, psm_cluster_membership.parquet) which
causes parquet_to_dat to emit empty/zero spectra; modify the logic that builds
parquet_files to filter out non-spectrum files by either (a) excluding known
metadata filenames (like 'cluster_metadata.parquet' and
'psm_cluster_membership.parquet') or (b) validating each candidate parquet file
contains the required columns ('mz_array' and 'intensity_array') before adding
it, so parquet_files only contains true spectrum parquet files passed to
parquet_to_dat.

209-214: ⚠️ Potential issue | 🟠 Major

Array length mismatch still not validated — previous comment not addressed.

If mz_array and intensity_array have different lengths for a row, _bin_spectrum will index past the shorter array (via np.argsort(-intensity_array) returning indices that may exceed len(mz_array)), which aborts the entire conversion instead of skipping a single bad spectrum.

Proposed fix
-                if len(mz_arr) == 0 or len(int_arr) == 0:
+                if len(mz_arr) == 0 or len(int_arr) == 0 or len(mz_arr) != len(int_arr):
                     skipped += 1
                     continue
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/maracluster_dat.py` around lines 209 - 214, After converting
mz_arr and int_arr to numpy arrays in the loop (the block that currently
contains mz_arr = np.asarray(...), int_arr = np.asarray(...)), add a strict
length-equality check for mz_arr and int_arr and treat mismatches like empty
arrays: increment skipped, optionally log a concise warning, and continue so
_bin_spectrum (and np.argsort on intensity_array) never sees misaligned lengths;
ensure this validation occurs before any indexing or calling
_bin_spectrum/intensity_array operations.

240-240: ⚠️ Potential issue | 🟠 Major

RT is still hard-coded to 0.0 — previous comment not addressed.

rt is listed in read_cols (line 161) and read from parquet, but the _pack_spectrum call here writes retentionTime=0.0. Any MaRaCluster mode consuming RT from the .dat payload will see all spectra at time zero.

Proposed fix
             runs = df['run_file_name'].values if 'run_file_name' in df.columns else [''] * len(df)
             peps = df['peptidoform'].values if 'peptidoform' in df.columns else [''] * len(df)
             scans = df['scan'].values if 'scan' in df.columns else [0] * len(df)
+            rts = df['rt'].values if 'rt' in df.columns else np.zeros(len(df), dtype=np.float32)
...
+                rt_val = rts[i]
+                rt_val = 0.0 if rt_val is None or (isinstance(rt_val, float) and np.isnan(rt_val)) else float(rt_val)
                 dat_buf.extend(_pack_spectrum(
-                    file_idx, scannr, charge, prec_mz, 0.0, peak_bins))
+                    file_idx, scannr, charge, prec_mz, rt_val, peak_bins))
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/maracluster_dat.py` at line 240, The call to _pack_spectrum is
hard-coding retentionTime to 0.0; replace that literal with the RT value read
from parquet (the variable named rt from read_cols) so retentionTime is passed
as rt (or the existing local RT variable) to _pack_spectrum; ensure the rt
variable is in scope at the call site and used instead of 0.0 when constructing
the MaRaCluster .dat payload.
🧹 Nitpick comments (10)
pyspectrafuse/commands/build_cluster_db.py (2)

235-235: Extraneous f prefix (Ruff F541).

f"Pass 2: Finding best PSM per cluster..." has no placeholders — drop the f.

-        logger.info(f"Pass 2: Finding best PSM per cluster...")
+        logger.info("Pass 2: Finding best PSM per cluster...")
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/build_cluster_db.py` at line 235, The log call uses an
unnecessary f-string; update the logger.info invocation that currently reads
logger.info(f"Pass 2: Finding best PSM per cluster...") to use a plain string
literal (logger.info("Pass 2: Finding best PSM per cluster..."))—locate the call
using the logger symbol in build_cluster_db.py (within the pass 2 / best PSM
selection section) and remove the leading f to eliminate the Ruff F541 warning.

146-153: Overly broad except Exception around DuckDB setup.

If any of the three SET statements fails (e.g., invalid memory_limit), the connection is closed and the original exception is re-raised — but that's what finally: conn.close() would do anyway if structured with the main try block. Consider merging this with the outer try/finally at line 202 instead of the duplicate close.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/build_cluster_db.py` around lines 146 - 153, The inner
try/except around the DuckDB SET statements (after conn = duckdb.connect(...))
is too broad and redundantly closes the connection; remove that except block and
let the existing outer try/finally that performs conn.close() manage cleanup,
i.e., execute the three SET statements (memory_limit, temp_directory using
tmp_dir, threads) directly after connect and rely on the outer finally to close
the connection; if you need to handle configuration-specific failures, catch a
more specific exception (e.g., duckdb.Error) around only the failing SET and
re-raise or log accordingly instead of closing the connection unconditionally.
pyspectrafuse/common/duckdb_ops.py (1)

20-21: Implicit Optional in temp_directory annotation (also at line 408).

PEP 484 prohibits implicit Optional with = None defaults. Same issue in scan_parquet_arrays signature. Minor style nit.

Proposed fix
-def _create_connection(memory_limit: str = '4GB', threads: int = 4,
-                       temp_directory: str = None) -> duckdb.DuckDBPyConnection:
+def _create_connection(memory_limit: str = '4GB', threads: int = 4,
+                       temp_directory: Optional[str] = None) -> duckdb.DuckDBPyConnection:
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/common/duckdb_ops.py` around lines 20 - 21, The function
signatures use a parameter default of None without an explicit Optional type;
update the annotations to use typing.Optional[str] for temp_directory in
_create_connection and for the analogous parameter in scan_parquet_arrays, and
add "from typing import Optional" to the imports so the signatures read e.g.
temp_directory: Optional[str] = None (apply the same change to the other
function).
pyspectrafuse/maracluster_dat.py (1)

366-407: split_dat_by_mz_window does 2× full I/O and a per-spectrum read loop.

Lines 374-378 read 100 bytes per spectrum in a Python loop just to extract precMz, then lines 406-407 re-read the entire file into memory. For a 50 GB .dat this is both slow and memory-heavy. Read once with np.frombuffer using a structured dtype (or a view over the bytes buffer) and slice by window.

Proposed refactor
-    # Read all spectra: extract precursor m/z from each 100-byte struct
-    dat_size = Path(dat_path).stat().st_size
-    n_spectra = dat_size // SPECTRUM_STRUCT.size
-    if n_spectra == 0:
-        return []
-
-    # Read all precursor m/z values from .dat (offset 12 in the struct: 3×uint32, then float32)
-    prec_mzs = np.zeros(n_spectra, dtype=np.float32)
-    with open(dat_path, 'rb') as f:
-        for i in range(n_spectra):
-            data = f.read(SPECTRUM_STRUCT.size)
-            unpacked = SPECTRUM_STRUCT.unpack(data)
-            prec_mzs[i] = unpacked[3]  # precMz is 4th field
-    ...
-    # Read all .dat records at once
-    with open(dat_path, 'rb') as f:
-        dat_data = f.read()
+    dat_size = Path(dat_path).stat().st_size
+    n_spectra = dat_size // SPECTRUM_STRUCT.size
+    if n_spectra == 0:
+        return []
+
+    with open(dat_path, 'rb') as f:
+        dat_data = f.read()
+    # precMz is the 4th field: offset 12 (3× uint32), dtype float32
+    prec_mzs = np.frombuffer(
+        dat_data, dtype=np.float32, count=n_spectra,
+        offset=0,
+    ).reshape(n_spectra, SPECTRUM_STRUCT.size // 4)[:, 3].copy()
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/maracluster_dat.py` around lines 366 - 407,
split_dat_by_mz_window currently does two full reads of the .dat file and
iterates per-spectrum with SPECTRUM_STRUCT.unpack to build prec_mzs, then
re-reads dat_path into dat_data; replace this with a single read and create a
numpy structured view (np.frombuffer or np.ndarray with the matching dtype) over
the bytes buffer so you can populate prec_mzs and index into raw records without
a per-spectrum Python loop. Specifically: read dat_path once into a bytes
object, use np.frombuffer(..., dtype=structured_dtype) or view on that buffer to
get the array of records, extract prec_mzs = records['precMz'] (or records[:,3])
directly, and then slice the buffer/array per-window instead of re-reading the
file; keep references to SPECTRUM_STRUCT only to derive the dtype if needed.
pyspectrafuse/incremental/resolve_clusters_dat.py (1)

110-111: Implicit reliance on NaN from left merge; make it explicit.

merged['is_rep'] == True / != True happen to produce the desired behavior for unmatched rows (where is_rep is NaN after the left merge → != True is True, i.e., treated as new spectra), but this is non-obvious and fragile. It also trips Ruff E712.

Proposed fix
-    rep_rows = merged[merged['is_rep'] == True].copy()
-    new_rows = merged[merged['is_rep'] != True].copy()
+    is_rep = merged['is_rep'].fillna(False).astype(bool)
+    rep_rows = merged[is_rep].copy()
+    new_rows = merged[~is_rep].copy()
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/incremental/resolve_clusters_dat.py` around lines 110 - 111,
The selection of representative vs new rows relies on NaN from the left merge
and uses equality comparisons to True (triggering Ruff E712); instead make the
NaN handling explicit by treating missing is_rep as False and then select rows
using boolean operations: for example, compute a clean boolean series like
merged['is_rep'].fillna(False) or merged['is_rep'].astype(bool) and use that to
build rep_rows and new_rows (e.g., rep_rows = merged[clean_is_rep] and new_rows
= merged[~clean_is_rep]); update any references to 'is_rep' selection in this
block accordingly.
pyspectrafuse/commands/spectrum2msp.py (2)

273-277: Silent except around cleanup swallows real errors.

conn.register('chunk_membership', ...) in DuckDB creates a temporary view; the proper teardown is conn.unregister('chunk_membership'). Also, catching bare Exception and pass hides genuine cleanup/state bugs (Ruff S110/BLE001). Prefer logging the exception at debug level so future regressions surface.

Proposed fix
-    # Unregister to allow re-registration in next chunk
-    try:
-        conn.execute("DROP VIEW IF EXISTS chunk_membership")
-    except Exception:
-        pass
+    # Unregister to allow re-registration in next chunk
+    try:
+        conn.unregister('chunk_membership')
+    except Exception as e:
+        logger.debug(f"chunk_membership unregister failed: {e}")
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/spectrum2msp.py` around lines 273 - 277, The cleanup
currently calls conn.execute("DROP VIEW IF EXISTS chunk_membership") inside a
bare try/except that swallows errors; replace this with
conn.unregister('chunk_membership') to properly tear down the temporary view
created by conn.register('chunk_membership'), and remove the bare except: if you
must catch exceptions, catch specific exceptions and log them at debug level
(using the module logger) instead of pass so failures during cleanup are visible
(refer to conn.execute, conn.unregister, conn.register, and the
'chunk_membership' view).

156-162: zip(scan_titles, dataset_names) silently truncates on mismatch.

dataset_names is built from scan_titles in the same loop above (Lines 335–338), so lengths will match in practice — but this is fragile. Consider using strict=True (Ruff B905) so any future refactor that desyncs these lists fails loudly instead of silently dropping title files.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/spectrum2msp.py` around lines 156 - 162, The loop
using zip(scan_titles_files, dataset_names) can silently drop items if the lists
differ; update the iteration in spectrum2msp.py to use zip(..., strict=True) (or
otherwise assert equal lengths before iterating) so mismatched lengths raise
immediately; specifically modify the block that builds titles_dfs (variables:
titles_dfs, scan_titles_files, dataset_names, and the call parse_scan_titles) to
iterate with strict=True to surface any future desynchronization.
pyspectrafuse/commands/incremental.py (3)

436-436: Drop the extraneous f prefix.

click.echo(f"Incremental merge complete:") has no placeholders (Ruff F541).

Proposed fix
-    click.echo(f"Incremental merge complete:")
+    click.echo("Incremental merge complete:")
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/incremental.py` at line 436, The call
click.echo(f"Incremental merge complete:") uses an unnecessary f-string; remove
the f prefix so it becomes click.echo("Incremental merge complete:") — find the
occurrence of click.echo(f"Incremental merge complete:") in
pyspectrafuse/commands/incremental.py (the exact string shown in the diff) and
replace it with a plain string literal.

265-267: Redundant defensive reset_index fallback.

meta_indexed = existing_meta.set_index('cluster_id') at Line 206/275 names the index cluster_id; reset_index() therefore always produces a cluster_id column, making the rename({'index': 'cluster_id'}) a no-op and the second if 'cluster_id' not in existing_meta.columns branch unreachable. Simplify for clarity. Same pattern at Lines 294–296.

Proposed fix
-            existing_meta = meta_indexed.reset_index().rename(columns={'index': 'cluster_id'})
-            if 'cluster_id' not in existing_meta.columns:
-                existing_meta = existing_meta.reset_index()
+            existing_meta = meta_indexed.reset_index()
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/incremental.py` around lines 265 - 267, The
conditional defensive reset is redundant: after meta_indexed =
existing_meta.set_index('cluster_id') the index is already named 'cluster_id',
so meta_indexed.reset_index().rename(columns={'index': 'cluster_id'}) always
yields a 'cluster_id' column and the subsequent if 'cluster_id' not in
existing_meta.columns branch is unreachable; replace that entire block with a
single existing_meta = meta_indexed.reset_index() (remove the rename and the
if/reset fallback), and apply the same simplification to the duplicate pattern
at the later occurrence (the block around lines 294–296) so both uses
consistently just call meta_indexed.reset_index().

258-260: Use .at[] for single-cell assignments for clarity and performance, though .loc[] works correctly here.

The current code with .loc[cid, col] actually functions correctly and does not raise the error mentioned in the original comment. With object-dtype columns (which these are), assigning a list to a single cell via .loc[scalar_index, column] works as intended and does not trigger the "Must have equal len keys and value" error—that error occurs in different assignment contexts (multi-row/column mismatches).

However, .at[] is still a valid best practice here:

  • Performance: .at is faster for single-cell operations
  • Clarity: .at explicitly signals single-cell intent, making code more maintainable

This applies to all instances at lines 258–260, 284–286, and 290–291.

Proposed refactoring
-                        meta_indexed.loc[cid, 'consensus_mz_array'] = row_out['mz_array'].astype(np.float32).tolist() if isinstance(row_out['mz_array'], np.ndarray) else list(row_out['mz_array'])
-                        meta_indexed.loc[cid, 'consensus_intensity_array'] = row_out['intensity_array'].astype(np.float32).tolist() if isinstance(row_out['intensity_array'], np.ndarray) else list(row_out['intensity_array'])
-                        meta_indexed.loc[cid, 'precursor_mz'] = float(row_out['pepmass'])
+                        meta_indexed.at[cid, 'consensus_mz_array'] = row_out['mz_array'].astype(np.float32).tolist() if isinstance(row_out['mz_array'], np.ndarray) else list(row_out['mz_array'])
+                        meta_indexed.at[cid, 'consensus_intensity_array'] = row_out['intensity_array'].astype(np.float32).tolist() if isinstance(row_out['intensity_array'], np.ndarray) else list(row_out['intensity_array'])
+                        meta_indexed.at[cid, 'precursor_mz'] = float(row_out['pepmass'])
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/incremental.py` around lines 258 - 260, Replace the
single-cell DataFrame assignments that use meta_indexed.loc[cid, ...]
(specifically the assignments to 'consensus_mz_array',
'consensus_intensity_array', and 'precursor_mz' in the incremental merging
logic) with meta_indexed.at[cid, ...] to express single-cell intent and gain a
small performance benefit; update all similar occurrences (the blocks that
assign row_out['mz_array'], row_out['intensity_array'], and
float(row_out['pepmass']) — also the analogous assignments around the other
occurrences referenced) so they use .at for scalar-index assignment instead of
.loc.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@pyspectrafuse/commands/build_cluster_db.py`:
- Around line 258-262: best_keys.merge with stats_purity may silently drop rows
if best_keys['cluster_id'] and stats_purity['cluster_id'] have different dtypes;
before calling best_keys = best_keys.merge(stats_purity, on='cluster_id',
how='left') ensure both columns share the same dtype (e.g. force both to string
via best_keys['cluster_id'] = best_keys['cluster_id'].astype(str) and
stats_purity['cluster_id'] = stats_purity['cluster_id'].astype(str)) or assert
their dtypes match and raise a clear error if not, so the left join does not
produce unexpected NaNs that are masked by the subsequent fillna calls.
- Around line 203-226: The SQL built in conn.execute uses f-strings that inject
user values (species, instrument, charge_int, psm_path, membership_tmp, etc.)
directly and will break on quotes or allow injection; change the code to build a
non-f-string SQL with parameter placeholders (use ? placeholders) and pass a
tuple of parameters to conn.execute so species and instrument are bound (e.g.
'... , ? AS species, ? AS instrument ... CAST(? AS TINYINT) AS charge ...') and
pass (species, instrument, charge_int, ...) as arguments; keep source_union (a
full SQL fragment) only if it is validated/sanitized or otherwise refactor to a
safe parameterized approach, and avoid inline f-strings for file names and
scalars—use parameters or properly validated/escaped values for membership_tmp
and psm_path as well.

In `@pyspectrafuse/commands/cluster2parquet.py`:
- Around line 52-54: The current call to find_target_ext_files(parquet_dir,
'.parquet') in cluster2parquet.py over-selects files (e.g.,
cluster_metadata.parquet and psm_cluster_membership.parquet) that are outputs
from prior runs and not valid PSM inputs, which breaks inject_cluster_info
downstream; update the logic after path_parquet_lst is built to filter out any
filenames matching 'cluster_metadata.parquet' and
'psm_cluster_membership.parquet' (or alternatively restrict to a stricter input
filename pattern) so only true PSM input parquet files are passed to
inject_cluster_info; adjust the variable path_parquet_lst accordingly and add a
clear error if no valid input files remain.

In `@pyspectrafuse/commands/incremental.py`:
- Around line 158-162: When output_dir is None the code currently resolves
out_membership/out_metadata to the source files
(existing_membership/existing_metadata) which allows append_psm_membership and
pq_write.write_table to overwrite originals in-place; change this so writes go
to temporary files in the same directory (e.g. create temp paths next to
existing_membership/existing_metadata), write to those temps, then atomically
swap them into place with os.replace; alternatively enforce that --output_dir
must differ from the parent of existing_* and error out. Update references in
incremental.py (output_dir, existing_membership, out_membership, out_metadata,
append_psm_membership, pq_write.write_table) to perform
temp-write-then-os.replace to avoid corruption on crash.

In `@pyspectrafuse/common/duckdb_ops.py`:
- Line 495: The loop using enumerate(zip(window_tsvs, window_mz_ranges)) can
silently truncate if the two lists differ in length; update the code to either
validate sizes before iterating (raise ValueError if len(window_tsvs) !=
len(window_mz_ranges)) or use zip(window_tsvs, window_mz_ranges, strict=True) in
the for statement (i.e., in the line defining for i, (tsv_path, (min_mz,
max_mz)) ...) so a mismatch raises immediately.

In `@pyspectrafuse/common/msp_utils.py`:
- Around line 79-87: The mz_intensity_val construction is using the raw
row['mz_array'] and never parsing intensity_array, causing iteration over
characters when arrays are stored as strings; update the logic in msp_utils.py
so both mz_arr (already parsed into mz_arr) and a parsed intensity_arr (use
ast.literal_eval if row['intensity_array'] is a str) are used, then build
mz_intensity_val by zipping mz_arr with intensity_arr (zip(mz_arr,
intensity_arr)) to produce "mz intensity" lines; keep existing num_peaks_val =
len(mz_arr) and ensure you reference mz_arr and intensity_arr variables rather
than row[...] when creating mz_intensity_val.

In `@pyspectrafuse/incremental/merge_results.py`:
- Around line 119-125: meta_df and stats_purity can have mismatched cluster_id
dtypes causing the left merge to produce all-NaN stats that are then masked by
fillna; before merging, explicitly coerce both sides to the same type (e.g.,
ensure meta_df['cluster_id'] and stats_purity['cluster_id'] are both cast to str
or both to int) so the meta_df.merge(stats_purity, on='cluster_id', how='left')
actually joins rows correctly and preserves computed stats/purity instead of
defaulting to fillna values.

In `@pyspectrafuse/incremental/representative_dat.py`:
- Around line 74-77: When charge_filter is provided but the input dataframe
lacks a 'charge' column, do not silently ignore it; detect this case and raise
an explicit exception. In the block where you currently check "if charge_filter
is not None and 'charge' in df.columns" (using variables charge_filter and df),
change logic to first test if charge_filter is not None and then if 'charge' is
missing raise a ValueError (or a domain-appropriate exception) explaining that
the metadata is missing the required 'charge' column; otherwise apply the
existing filter df = df[df['charge'] == charge_filter] and keep the df.empty
continue behavior. Ensure the error message references "charge_filter" and
"charge" so callers understand the mismatch.

In `@pyspectrafuse/maracluster_dat.py`:
- Around line 221-228: The current precursor m/z fallback can still produce NaN
and produce a NaN prec_mass that breaks filtering; in the block that computes
prec_mz/prec_mass (variables prec_mz, prec_mass using obs_mz, calc_mz, charge
and PROTON_MASS) detect when both observed_mz[i] and calculated_mz[i] are
missing/NaN and bail out by skipping this spectrum (e.g., continue) before
calling _bin_spectrum, rather than converting NaN to float and proceeding;
ensure you use numpy.isnan to test calc_mz[i] and observed_mz[i] and only
compute prec_mass when prec_mz is a valid finite number.

In `@tests/test_incremental.py`:
- Around line 141-144: The failing test asserts None against a pandas-produced
NaN; update the assertion in tests/test_incremental.py to use pandas' isna check
instead of identity comparison: replace the "old_cluster_id is None" check with
pd.isna(df.iloc[2]['old_cluster_id']) and ensure pandas is imported as pd in the
test file; keep the existing truth checks (e.g., df.iloc[0]['is_rep'] == True)
unchanged or use == True/== False per style guidance.

---

Duplicate comments:
In `@pyspectrafuse/maracluster_dat.py`:
- Around line 290-292: The current use of find_target_ext_files(parquet_dir,
'.parquet') collects non-spectrum parquet files (e.g., cluster_metadata.parquet,
psm_cluster_membership.parquet) which causes parquet_to_dat to emit empty/zero
spectra; modify the logic that builds parquet_files to filter out non-spectrum
files by either (a) excluding known metadata filenames (like
'cluster_metadata.parquet' and 'psm_cluster_membership.parquet') or (b)
validating each candidate parquet file contains the required columns ('mz_array'
and 'intensity_array') before adding it, so parquet_files only contains true
spectrum parquet files passed to parquet_to_dat.
- Around line 209-214: After converting mz_arr and int_arr to numpy arrays in
the loop (the block that currently contains mz_arr = np.asarray(...), int_arr =
np.asarray(...)), add a strict length-equality check for mz_arr and int_arr and
treat mismatches like empty arrays: increment skipped, optionally log a concise
warning, and continue so _bin_spectrum (and np.argsort on intensity_array) never
sees misaligned lengths; ensure this validation occurs before any indexing or
calling _bin_spectrum/intensity_array operations.
- Line 240: The call to _pack_spectrum is hard-coding retentionTime to 0.0;
replace that literal with the RT value read from parquet (the variable named rt
from read_cols) so retentionTime is passed as rt (or the existing local RT
variable) to _pack_spectrum; ensure the rt variable is in scope at the call site
and used instead of 0.0 when constructing the MaRaCluster .dat payload.

---

Nitpick comments:
In `@pyspectrafuse/commands/build_cluster_db.py`:
- Line 235: The log call uses an unnecessary f-string; update the logger.info
invocation that currently reads logger.info(f"Pass 2: Finding best PSM per
cluster...") to use a plain string literal (logger.info("Pass 2: Finding best
PSM per cluster..."))—locate the call using the logger symbol in
build_cluster_db.py (within the pass 2 / best PSM selection section) and remove
the leading f to eliminate the Ruff F541 warning.
- Around line 146-153: The inner try/except around the DuckDB SET statements
(after conn = duckdb.connect(...)) is too broad and redundantly closes the
connection; remove that except block and let the existing outer try/finally that
performs conn.close() manage cleanup, i.e., execute the three SET statements
(memory_limit, temp_directory using tmp_dir, threads) directly after connect and
rely on the outer finally to close the connection; if you need to handle
configuration-specific failures, catch a more specific exception (e.g.,
duckdb.Error) around only the failing SET and re-raise or log accordingly
instead of closing the connection unconditionally.

In `@pyspectrafuse/commands/incremental.py`:
- Line 436: The call click.echo(f"Incremental merge complete:") uses an
unnecessary f-string; remove the f prefix so it becomes click.echo("Incremental
merge complete:") — find the occurrence of click.echo(f"Incremental merge
complete:") in pyspectrafuse/commands/incremental.py (the exact string shown in
the diff) and replace it with a plain string literal.
- Around line 265-267: The conditional defensive reset is redundant: after
meta_indexed = existing_meta.set_index('cluster_id') the index is already named
'cluster_id', so meta_indexed.reset_index().rename(columns={'index':
'cluster_id'}) always yields a 'cluster_id' column and the subsequent if
'cluster_id' not in existing_meta.columns branch is unreachable; replace that
entire block with a single existing_meta = meta_indexed.reset_index() (remove
the rename and the if/reset fallback), and apply the same simplification to the
duplicate pattern at the later occurrence (the block around lines 294–296) so
both uses consistently just call meta_indexed.reset_index().
- Around line 258-260: Replace the single-cell DataFrame assignments that use
meta_indexed.loc[cid, ...] (specifically the assignments to
'consensus_mz_array', 'consensus_intensity_array', and 'precursor_mz' in the
incremental merging logic) with meta_indexed.at[cid, ...] to express single-cell
intent and gain a small performance benefit; update all similar occurrences (the
blocks that assign row_out['mz_array'], row_out['intensity_array'], and
float(row_out['pepmass']) — also the analogous assignments around the other
occurrences referenced) so they use .at for scalar-index assignment instead of
.loc.

In `@pyspectrafuse/commands/spectrum2msp.py`:
- Around line 273-277: The cleanup currently calls conn.execute("DROP VIEW IF
EXISTS chunk_membership") inside a bare try/except that swallows errors; replace
this with conn.unregister('chunk_membership') to properly tear down the
temporary view created by conn.register('chunk_membership'), and remove the bare
except: if you must catch exceptions, catch specific exceptions and log them at
debug level (using the module logger) instead of pass so failures during cleanup
are visible (refer to conn.execute, conn.unregister, conn.register, and the
'chunk_membership' view).
- Around line 156-162: The loop using zip(scan_titles_files, dataset_names) can
silently drop items if the lists differ; update the iteration in spectrum2msp.py
to use zip(..., strict=True) (or otherwise assert equal lengths before
iterating) so mismatched lengths raise immediately; specifically modify the
block that builds titles_dfs (variables: titles_dfs, scan_titles_files,
dataset_names, and the call parse_scan_titles) to iterate with strict=True to
surface any future desynchronization.

In `@pyspectrafuse/common/duckdb_ops.py`:
- Around line 20-21: The function signatures use a parameter default of None
without an explicit Optional type; update the annotations to use
typing.Optional[str] for temp_directory in _create_connection and for the
analogous parameter in scan_parquet_arrays, and add "from typing import
Optional" to the imports so the signatures read e.g. temp_directory:
Optional[str] = None (apply the same change to the other function).

In `@pyspectrafuse/incremental/resolve_clusters_dat.py`:
- Around line 110-111: The selection of representative vs new rows relies on NaN
from the left merge and uses equality comparisons to True (triggering Ruff
E712); instead make the NaN handling explicit by treating missing is_rep as
False and then select rows using boolean operations: for example, compute a
clean boolean series like merged['is_rep'].fillna(False) or
merged['is_rep'].astype(bool) and use that to build rep_rows and new_rows (e.g.,
rep_rows = merged[clean_is_rep] and new_rows = merged[~clean_is_rep]); update
any references to 'is_rep' selection in this block accordingly.

In `@pyspectrafuse/maracluster_dat.py`:
- Around line 366-407: split_dat_by_mz_window currently does two full reads of
the .dat file and iterates per-spectrum with SPECTRUM_STRUCT.unpack to build
prec_mzs, then re-reads dat_path into dat_data; replace this with a single read
and create a numpy structured view (np.frombuffer or np.ndarray with the
matching dtype) over the bytes buffer so you can populate prec_mzs and index
into raw records without a per-spectrum Python loop. Specifically: read dat_path
once into a bytes object, use np.frombuffer(..., dtype=structured_dtype) or view
on that buffer to get the array of records, extract prec_mzs = records['precMz']
(or records[:,3]) directly, and then slice the buffer/array per-window instead
of re-reading the file; keep references to SPECTRUM_STRUCT only to derive the
dtype if needed.
🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

  • Push a commit to this branch (recommended)
  • Create a new PR with the fixes

ℹ️ Review info
⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 6c944b0b-5297-4d91-8721-e5e827c67707

📥 Commits

Reviewing files that changed from the base of the PR and between a2fa10a and 3229e22.

📒 Files selected for processing (26)
  • Dockerfile
  • README.md
  • pyproject.toml
  • pyspectrafuse/__init__.py
  • pyspectrafuse/commands/build_cluster_db.py
  • pyspectrafuse/commands/cluster2parquet.py
  • pyspectrafuse/commands/incremental.py
  • pyspectrafuse/commands/quantmsio2mgf.py
  • pyspectrafuse/commands/spectrum2msp.py
  • pyspectrafuse/common/__init__.py
  • pyspectrafuse/common/duckdb_ops.py
  • pyspectrafuse/common/msp_utils.py
  • pyspectrafuse/common/parquet_utils.py
  • pyspectrafuse/common/schemas.py
  • pyspectrafuse/consensus_strategy/__init__.py
  • pyspectrafuse/consensus_strategy/average_spectrum_strategy.py
  • pyspectrafuse/incremental/merge_results.py
  • pyspectrafuse/incremental/representative_dat.py
  • pyspectrafuse/incremental/resolve_clusters_dat.py
  • pyspectrafuse/maracluster_dat.py
  • pyspectrafuse/mgf_convert/parquet2mgf.py
  • pyspectrafuse/pyspectrafuse.py
  • requirements.txt
  • tests/test_cli.py
  • tests/test_incremental.py
  • tests/test_integration_ftp.py
💤 Files with no reviewable changes (2)
  • pyspectrafuse/commands/quantmsio2mgf.py
  • pyspectrafuse/mgf_convert/parquet2mgf.py
✅ Files skipped from review due to trivial changes (5)
  • requirements.txt
  • pyspectrafuse/common/init.py
  • pyspectrafuse/init.py
  • pyspectrafuse/consensus_strategy/init.py
  • pyspectrafuse/common/schemas.py
🚧 Files skipped from review as they are similar to previous changes (3)
  • pyspectrafuse/consensus_strategy/average_spectrum_strategy.py
  • pyspectrafuse/common/parquet_utils.py
  • pyproject.toml

Comment on lines +203 to +226
conn.execute(f"""
COPY (
SELECT
m.cluster_id,
'mzspec:' || m.dataset || ':' || m.run_file_name || ':scan:'
|| CAST(m.orig_scan AS VARCHAR) || ':' || m.peptidoform
|| '/' || CAST({charge_int} AS VARCHAR) AS usi,
m.dataset AS project_accession,
m.run_file_name AS reference_file_name,
CAST(m.orig_scan AS INTEGER) AS scan,
m.peptidoform,
CAST({charge_int} AS TINYINT) AS charge,
COALESCE(s.precursor_mz, 0.0) AS precursor_mz,
s.posterior_error_probability,
s.global_qvalue,
'{species}' AS species,
'{instrument}' AS instrument
FROM read_parquet('{membership_tmp}') m
LEFT JOIN ({source_union}) s
ON m.run_file_name = s.run_file_name
AND m.orig_scan = s.orig_scan
ORDER BY m.cluster_id
) TO '{psm_path}' (FORMAT PARQUET, COMPRESSION ZSTD)
""")

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

String-interpolated {species} / {instrument} will break on any single quote.

User-supplied species and instrument are dropped straight into the SQL string. A value containing an apostrophe (or backslash) produces malformed SQL and fails the whole COPY. Pass them as parameters or escape. Same concern applies to line 235 (f-string with no placeholders) and to the {charge_int}/{psm_path} interpolations throughout — scalars should use DuckDB parameters.

Proposed fix
         conn.execute(f"""
             COPY (
                 SELECT
                     m.cluster_id,
                     'mzspec:' || m.dataset || ':' || m.run_file_name || ':scan:'
                         || CAST(m.orig_scan AS VARCHAR) || ':' || m.peptidoform
-                        || '/' || CAST({charge_int} AS VARCHAR) AS usi,
+                        || '/' || CAST(? AS VARCHAR) AS usi,
                     m.dataset AS project_accession,
                     m.run_file_name AS reference_file_name,
                     CAST(m.orig_scan AS INTEGER) AS scan,
                     m.peptidoform,
-                    CAST({charge_int} AS TINYINT) AS charge,
+                    CAST(? AS TINYINT) AS charge,
                     COALESCE(s.precursor_mz, 0.0) AS precursor_mz,
                     s.posterior_error_probability,
                     s.global_qvalue,
-                    '{species}' AS species,
-                    '{instrument}' AS instrument
+                    ? AS species,
+                    ? AS instrument
                 FROM read_parquet('{membership_tmp}') m
                 LEFT JOIN ({source_union}) s
                     ON m.run_file_name = s.run_file_name
                     AND m.orig_scan = s.orig_scan
                 ORDER BY m.cluster_id
             ) TO '{psm_path}' (FORMAT PARQUET, COMPRESSION ZSTD)
-        """)
+        """, [charge_int, charge_int, species, instrument])
🧰 Tools
🪛 Ruff (0.15.10)

[error] 203-226: Possible SQL injection vector through string-based query construction

(S608)

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/build_cluster_db.py` around lines 203 - 226, The SQL
built in conn.execute uses f-strings that inject user values (species,
instrument, charge_int, psm_path, membership_tmp, etc.) directly and will break
on quotes or allow injection; change the code to build a non-f-string SQL with
parameter placeholders (use ? placeholders) and pass a tuple of parameters to
conn.execute so species and instrument are bound (e.g. '... , ? AS species, ? AS
instrument ... CAST(? AS TINYINT) AS charge ...') and pass (species, instrument,
charge_int, ...) as arguments; keep source_union (a full SQL fragment) only if
it is validated/sanitized or otherwise refactor to a safe parameterized
approach, and avoid inline f-strings for file names and scalars—use parameters
or properly validated/escaped values for membership_tmp and psm_path as well.

Comment on lines +258 to +262
best_keys = best_keys.merge(stats_purity, on='cluster_id', how='left')
best_keys['purity'] = best_keys['purity'].fillna(1.0).astype(np.float32)
best_keys['member_count'] = best_keys['member_count'].fillna(1).astype(np.int32)
best_keys['project_count'] = best_keys['project_count'].fillna(1).astype(np.int16)
del stats_purity; gc.collect()

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Same cluster_id dtype caveat as in merge_results.py.

best_keys['cluster_id'] comes from a DuckDB read (so likely object/string from the parquet), while stats_purity['cluster_id'] also comes from DuckDB. They should match in this file because they both came from the same psm_cluster_membership.parquet, but if schema evolution ever introduces a mismatch, the left merge silently turns into NaN and fillna(1)/fillna(1.0) masks the failure. Consider forcing astype(str) on both before merge, or asserting equal dtypes.

🧰 Tools
🪛 Ruff (0.15.10)

[error] 262-262: Multiple statements on one line (semicolon)

(E702)

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/build_cluster_db.py` around lines 258 - 262,
best_keys.merge with stats_purity may silently drop rows if
best_keys['cluster_id'] and stats_purity['cluster_id'] have different dtypes;
before calling best_keys = best_keys.merge(stats_purity, on='cluster_id',
how='left') ensure both columns share the same dtype (e.g. force both to string
via best_keys['cluster_id'] = best_keys['cluster_id'].astype(str) and
stats_purity['cluster_id'] = stats_purity['cluster_id'].astype(str)) or assert
their dtypes match and raise a clear error if not, so the left join does not
produce unexpected NaNs that are masked by the subsequent fillna calls.

Comment thread pyspectrafuse/commands/incremental.py Outdated
Comment on lines +158 to +162
if output_dir is None:
output_dir = str(Path(existing_membership).parent)
Path(output_dir).mkdir(parents=True, exist_ok=True)

out_membership = str(Path(output_dir) / 'psm_cluster_membership.parquet')

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Default output path overwrites existing metadata/membership in place — risk of corruption on failure.

When --output_dir is omitted, out_membership and out_metadata resolve to the same paths as existing_membership / existing_metadata. append_psm_membership and the later pq_write.write_table(table, out_metadata, ...) will read the source and overwrite it. Any crash mid-write leaves the cluster DB in a torn state with no rollback path. Write to a temp file and os.replace atomically, or require a distinct --output_dir.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/incremental.py` around lines 158 - 162, When
output_dir is None the code currently resolves out_membership/out_metadata to
the source files (existing_membership/existing_metadata) which allows
append_psm_membership and pq_write.write_table to overwrite originals in-place;
change this so writes go to temporary files in the same directory (e.g. create
temp paths next to existing_membership/existing_metadata), write to those temps,
then atomically swap them into place with os.replace; alternatively enforce that
--output_dir must differ from the parent of existing_* and error out. Update
references in incremental.py (output_dir, existing_membership, out_membership,
out_metadata, append_psm_membership, pq_write.write_table) to perform
temp-write-then-os.replace to avoid corruption on crash.

Comment thread pyspectrafuse/common/duckdb_ops.py Resolved
Comment on lines +79 to +87
mz_arr = row['mz_array']
if isinstance(mz_arr, str):
mz_arr = ast.literal_eval(mz_arr)
num_peaks_val = len(mz_arr)
comment_val = f'clusterID={MspUtil().usi_to_uuid(row["usi"])} Nreps={row["Nreps"]} PEP={row["posterior_error_probability"]}'
mz_intensity_val = Parquet2Mgf.get_mz_intensity_str(row['mz_array'], row['intensity_array'])
mz_intensity_val = '\n'.join(
f'{mz} {intensity}'
for mz, intensity in zip(row['mz_array'], row['intensity_array'])
)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

Bug: mz_intensity_val uses the raw (possibly string) row['mz_array'] instead of the parsed mz_arr, and never parses intensity_array.

When row['mz_array'] is a string (per the existing test in tests/test_msp_utils.py which stringifies arrays), you parse it into mz_arr and use that only for num_peaks_val. The zip(row['mz_array'], row['intensity_array']) at Line 86 then iterates over the characters of the unparsed strings, producing garbage peak lines like [ [ / 0 0. Additionally, row['intensity_array'] is never converted even though it can arrive as a string through the same path.

🐛 Proposed fix
     mz_arr = row['mz_array']
     if isinstance(mz_arr, str):
         mz_arr = ast.literal_eval(mz_arr)
+    intensity_arr = row['intensity_array']
+    if isinstance(intensity_arr, str):
+        intensity_arr = ast.literal_eval(intensity_arr)
     num_peaks_val = len(mz_arr)
     comment_val = f'clusterID={MspUtil().usi_to_uuid(row["usi"])} Nreps={row["Nreps"]} PEP={row["posterior_error_probability"]}'
     mz_intensity_val = '\n'.join(
         f'{mz} {intensity}'
-        for mz, intensity in zip(row['mz_array'], row['intensity_array'])
+        for mz, intensity in zip(mz_arr, intensity_arr)
     )
📝 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.

Suggested change
mz_arr = row['mz_array']
if isinstance(mz_arr, str):
mz_arr = ast.literal_eval(mz_arr)
num_peaks_val = len(mz_arr)
comment_val = f'clusterID={MspUtil().usi_to_uuid(row["usi"])} Nreps={row["Nreps"]} PEP={row["posterior_error_probability"]}'
mz_intensity_val = Parquet2Mgf.get_mz_intensity_str(row['mz_array'], row['intensity_array'])
mz_intensity_val = '\n'.join(
f'{mz} {intensity}'
for mz, intensity in zip(row['mz_array'], row['intensity_array'])
)
mz_arr = row['mz_array']
if isinstance(mz_arr, str):
mz_arr = ast.literal_eval(mz_arr)
intensity_arr = row['intensity_array']
if isinstance(intensity_arr, str):
intensity_arr = ast.literal_eval(intensity_arr)
num_peaks_val = len(mz_arr)
comment_val = f'clusterID={MspUtil().usi_to_uuid(row["usi"])} Nreps={row["Nreps"]} PEP={row["posterior_error_probability"]}'
mz_intensity_val = '\n'.join(
f'{mz} {intensity}'
for mz, intensity in zip(mz_arr, intensity_arr)
)
🧰 Tools
🪛 Ruff (0.15.10)

[warning] 86-86: zip() without an explicit strict= parameter

Add explicit value for parameter strict=

(B905)

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/common/msp_utils.py` around lines 79 - 87, The mz_intensity_val
construction is using the raw row['mz_array'] and never parsing intensity_array,
causing iteration over characters when arrays are stored as strings; update the
logic in msp_utils.py so both mz_arr (already parsed into mz_arr) and a parsed
intensity_arr (use ast.literal_eval if row['intensity_array'] is a str) are
used, then build mz_intensity_val by zipping mz_arr with intensity_arr
(zip(mz_arr, intensity_arr)) to produce "mz intensity" lines; keep existing
num_peaks_val = len(mz_arr) and ensure you reference mz_arr and intensity_arr
variables rather than row[...] when creating mz_intensity_val.

Comment on lines +74 to +77
if charge_filter is not None and 'charge' in df.columns:
df = df[df['charge'] == charge_filter]
if df.empty:
continue

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Silent no-op when charge_filter is set but the charge column is missing.

If a caller supplies charge_filter against a metadata file lacking the charge column, the filter is silently ignored and all clusters are emitted. Since the CLI advertises this as a strict filter, prefer raising instead.

Proposed fix
-            if charge_filter is not None and 'charge' in df.columns:
-                df = df[df['charge'] == charge_filter]
-                if df.empty:
-                    continue
+            if charge_filter is not None:
+                if 'charge' not in df.columns:
+                    raise ValueError(
+                        "charge_filter requested but 'charge' column is missing "
+                        f"from {cluster_metadata_path}")
+                df = df[df['charge'] == charge_filter]
+                if df.empty:
+                    continue
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/incremental/representative_dat.py` around lines 74 - 77, When
charge_filter is provided but the input dataframe lacks a 'charge' column, do
not silently ignore it; detect this case and raise an explicit exception. In the
block where you currently check "if charge_filter is not None and 'charge' in
df.columns" (using variables charge_filter and df), change logic to first test
if charge_filter is not None and then if 'charge' is missing raise a ValueError
(or a domain-appropriate exception) explaining that the metadata is missing the
required 'charge' column; otherwise apply the existing filter df =
df[df['charge'] == charge_filter] and keep the df.empty continue behavior.
Ensure the error message references "charge_filter" and "charge" so callers
understand the mismatch.

Comment on lines +221 to +228
# Precursor m/z: prefer observed_mz, fall back to calculated_mz
prec_mz = obs_mz[i] if obs_mz is not None else None
if prec_mz is None or np.isnan(prec_mz):
prec_mz = calc_mz[i] if calc_mz is not None else 0.0
prec_mz = float(prec_mz)

# Neutral mass for peak filtering
prec_mass = prec_mz * charge - PROTON_MASS * (charge - 1)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

calc_mz fallback may still yield NaN and pollute prec_mass.

If observed_mz[i] is NaN/None and calculated_mz[i] is also NaN, prec_mz becomes NaN after float(), then prec_mass becomes NaN, and _bin_spectrum's mz >= prec_mass comparison evaluates False for every peak — so you silently retain all peaks regardless of the precursor filter. Skip the spectrum instead.

Proposed fix
                 prec_mz = obs_mz[i] if obs_mz is not None else None
                 if prec_mz is None or np.isnan(prec_mz):
                     prec_mz = calc_mz[i] if calc_mz is not None else 0.0
-                prec_mz = float(prec_mz)
+                if prec_mz is None or np.isnan(prec_mz) or prec_mz <= 0:
+                    skipped += 1
+                    continue
+                prec_mz = float(prec_mz)
📝 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.

Suggested change
# Precursor m/z: prefer observed_mz, fall back to calculated_mz
prec_mz = obs_mz[i] if obs_mz is not None else None
if prec_mz is None or np.isnan(prec_mz):
prec_mz = calc_mz[i] if calc_mz is not None else 0.0
prec_mz = float(prec_mz)
# Neutral mass for peak filtering
prec_mass = prec_mz * charge - PROTON_MASS * (charge - 1)
# Precursor m/z: prefer observed_mz, fall back to calculated_mz
prec_mz = obs_mz[i] if obs_mz is not None else None
if prec_mz is None or np.isnan(prec_mz):
prec_mz = calc_mz[i] if calc_mz is not None else 0.0
if prec_mz is None or np.isnan(prec_mz) or prec_mz <= 0:
skipped += 1
continue
prec_mz = float(prec_mz)
# Neutral mass for peak filtering
prec_mass = prec_mz * charge - PROTON_MASS * (charge - 1)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/maracluster_dat.py` around lines 221 - 228, The current
precursor m/z fallback can still produce NaN and produce a NaN prec_mass that
breaks filtering; in the block that computes prec_mz/prec_mass (variables
prec_mz, prec_mass using obs_mz, calc_mz, charge and PROTON_MASS) detect when
both observed_mz[i] and calculated_mz[i] are missing/NaN and bail out by
skipping this spectrum (e.g., continue) before calling _bin_spectrum, rather
than converting NaN to float and proceeding; ensure you use numpy.isnan to test
calc_mz[i] and observed_mz[i] and only compute prec_mass when prec_mz is a valid
finite number.

Comment thread tests/test_incremental.py
Comment on lines +141 to +144
assert df.iloc[0]['is_rep'] == True
assert df.iloc[0]['old_cluster_id'] == 'CLUS_001'
assert df.iloc[2]['is_rep'] == False
assert df.iloc[2]['old_cluster_id'] is None

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

Test fails in CI: old_cluster_id is None but pandas returns NaN.

The pipeline reports assert nan is None at Line 144. When pandas constructs the column from mixed str/None values, the None becomes NaN (or np.nan), which is not None. Use pd.isna(...) for the NaN check and == True/== False will still work but bool(...) / pd.isna is cleaner per Ruff E712.

Proposed fix
-        assert df.iloc[0]['is_rep'] == True
+        assert bool(df.iloc[0]['is_rep']) is True
         assert df.iloc[0]['old_cluster_id'] == 'CLUS_001'
-        assert df.iloc[2]['is_rep'] == False
-        assert df.iloc[2]['old_cluster_id'] is None
+        assert bool(df.iloc[2]['is_rep']) is False
+        assert pd.isna(df.iloc[2]['old_cluster_id'])
🧰 Tools
🪛 GitHub Actions: Tests

[error] 144-144: Pytest failure in TestParseScanTitles.test_basic_parsing: expected df.iloc[2]['old_cluster_id'] is None, but got NaN ('assert nan is None').

🪛 Ruff (0.15.10)

[error] 141-141: Avoid equality comparisons to True; use df.iloc[0]['is_rep']: for truth checks

Replace with df.iloc[0]['is_rep']

(E712)


[error] 143-143: Avoid equality comparisons to False; use not df.iloc[2]['is_rep']: for false checks

Replace with not df.iloc[2]['is_rep']

(E712)

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@tests/test_incremental.py` around lines 141 - 144, The failing test asserts
None against a pandas-produced NaN; update the assertion in
tests/test_incremental.py to use pandas' isna check instead of identity
comparison: replace the "old_cluster_id is None" check with
pd.isna(df.iloc[2]['old_cluster_id']) and ensure pandas is imported as pd in the
test file; keep the existing truth checks (e.g., df.iloc[0]['is_rep'] == True)
unchanged or use == True/== False per style guidance.

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actionable comments posted: 3

🧹 Nitpick comments (1)
.github/workflows/tests.yml (1)

21-28: Enable uv cache in setup-uv for faster CI runs.

astral-sh/setup-uv@v4 supports built-in caching keyed on uv.lock/pyproject.toml, which significantly speeds up CI. Since pyproject.toml exists with test extras defined, uv sync --extra test will work correctly (uv automatically generates a lockfile if one is missing).

♻️ Suggested tweak
    - name: Install uv
      uses: astral-sh/setup-uv@v4
+      with:
+        enable-cache: true
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In @.github/workflows/tests.yml around lines 21 - 28, Update the "Install uv"
step that uses astral-sh/setup-uv@v4 to enable the built-in cache by adding a
with block such as "with: cache: true" so the action keys caching against
uv.lock/pyproject.toml; this will make subsequent "uv python install ${{
matrix.python-version }}" and "uv sync --extra test" runs faster by reusing the
cached environment (refer to the step using astral-sh/setup-uv@v4 and the
commands uv python install and uv sync --extra test when making the change).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.

Inline comments:
In @.github/workflows/pyspectrafuse-containers.yml:
- Around line 94-96: The workflow uses the deprecated action uses:
eWaterCycle/setup-singularity@v7 and the singularity-version input; replace that
with the maintained action uses: eWaterCycle/setup-apptainer@v2 and swap the
singularity-version key for apptainer-version (for example apptainer-version:
1.3.6) so the job uses Apptainer instead of the deprecated Singularity action.
- Around line 46-49: Update the GitHub Actions steps that use docker actions to
their latest stable tags: replace docker/setup-buildx-action@v3 with
docker/setup-buildx-action@v4, docker/login-action@v3 with
docker/login-action@v4, and any docker/build-push-action@v? with
docker/build-push-action@v7; update the action references in the workflow to
those exact tags (e.g., in the steps using docker/setup-buildx-action,
docker/login-action, and docker/build-push-action) and run a quick workflow lint
or dry-run to confirm no breaking input changes are required by the newer major
versions.
- Around line 79-85: The build step using docker/build-push-action@v6 currently
sets push: false for PRs but doesn't set load, so the image isn't loaded into
the local Docker daemon and subsequent docker run tests fail; update the "Build
and Push pyspectrafuse image" action to add a load parameter (e.g., load: ${{
github.event_name == 'pull_request' }} or load: true for PRs) so that when push
is false on pull_request the built image is loaded locally, ensuring the later
docker run tests (the commands invoking ghcr.io/bigbio/pyspectrafuse:${{
steps.version.outputs.VERSION }} and the Python import checks) can run against
the locally loaded image while keeping platforms: linux/amd64 and the existing
push conditional logic unchanged.

---

Nitpick comments:
In @.github/workflows/tests.yml:
- Around line 21-28: Update the "Install uv" step that uses
astral-sh/setup-uv@v4 to enable the built-in cache by adding a with block such
as "with: cache: true" so the action keys caching against
uv.lock/pyproject.toml; this will make subsequent "uv python install ${{
matrix.python-version }}" and "uv sync --extra test" runs faster by reusing the
cached environment (refer to the step using astral-sh/setup-uv@v4 and the
commands uv python install and uv sync --extra test when making the change).
🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

  • Push a commit to this branch (recommended)
  • Create a new PR with the fixes

ℹ️ Review info
⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 47973ded-5ce2-4033-8944-ddd112ef9a7e

📥 Commits

Reviewing files that changed from the base of the PR and between 3229e22 and 4628df8.

📒 Files selected for processing (2)
  • .github/workflows/pyspectrafuse-containers.yml
  • .github/workflows/tests.yml

Comment on lines +46 to +49
uses: docker/setup-buildx-action@v3

- name: Log in to GitHub Container Registry
uses: docker/login-action@5e57cd118135c172c3672efd75eb46360885c0ef
uses: docker/login-action@v3

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

🧩 Analysis chain

🌐 Web query:

Latest stable major versions of docker/setup-buildx-action, docker/login-action, and docker/build-push-action in 2026

💡 Result:

Latest stable major versions as of 2026-04-17: - docker/setup-buildx-action: v4 (released 2026-03-05) - docker/login-action: v4 (released 2026-04-02) - docker/build-push-action: v7 (released 2026-03-05)

Citations:


Update Docker action versions to latest stable releases. The current versions are outdated: docker/setup-buildx-action@v4, docker/login-action@v4, and docker/build-push-action@v7 are available as of April 2026.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In @.github/workflows/pyspectrafuse-containers.yml around lines 46 - 49, Update
the GitHub Actions steps that use docker actions to their latest stable tags:
replace docker/setup-buildx-action@v3 with docker/setup-buildx-action@v4,
docker/login-action@v3 with docker/login-action@v4, and any
docker/build-push-action@v? with docker/build-push-action@v7; update the action
references in the workflow to those exact tags (e.g., in the steps using
docker/setup-buildx-action, docker/login-action, and docker/build-push-action)
and run a quick workflow lint or dry-run to confirm no breaking input changes
are required by the newer major versions.

Comment on lines +79 to 85
echo "Testing CLI..."
docker run --rm ghcr.io/bigbio/pyspectrafuse:${{ steps.version.outputs.VERSION }} --help
echo "Testing Python imports..."
docker run --rm --entrypoint python ghcr.io/bigbio/pyspectrafuse:${{ steps.version.outputs.VERSION }} -c "from pyspectrafuse.pyspectrafuse import main; print('CLI import successful')" || docker run --rm --entrypoint python ghcr.io/bigbio/pyspectrafuse:latest -c "from pyspectrafuse.pyspectrafuse import main; print('CLI import successful')"
echo "Testing consensus strategy imports..."
docker run --rm --entrypoint python ghcr.io/bigbio/pyspectrafuse:${{ steps.version.outputs.VERSION }} -c "from pyspectrafuse.consensus_strategy.best_spetrum_strategy import BestSpectrumStrategy; print('BestSpectrumStrategy import successful')" || docker run --rm --entrypoint python ghcr.io/bigbio/pyspectrafuse:latest -c "from pyspectrafuse.consensus_strategy.best_spetrum_strategy import BestSpectrumStrategy; print('BestSpectrumStrategy import successful')"
echo "Testing spectrum_utils import (numba caching test)..."
docker run --rm --entrypoint python ghcr.io/bigbio/pyspectrafuse:${{ steps.version.outputs.VERSION }} -c "import spectrum_utils.spectrum as sus; print('spectrum_utils import successful')" || docker run --rm --entrypoint python ghcr.io/bigbio/pyspectrafuse:latest -c "import spectrum_utils.spectrum as sus; print('spectrum_utils import successful')"
docker run --rm --entrypoint python ghcr.io/bigbio/pyspectrafuse:${{ steps.version.outputs.VERSION }} -c "from pyspectrafuse.pyspectrafuse import main; print('CLI import OK')"
echo "Testing dat conversion import..."
docker run --rm --entrypoint python ghcr.io/bigbio/pyspectrafuse:${{ steps.version.outputs.VERSION }} -c "from pyspectrafuse.maracluster_dat import parquet_to_dat; print('dat conversion import OK')"
echo "All container tests passed!"

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🌐 Web query:

docker/build-push-action v6 load image into local docker daemon when push is false

💡 Result:

To load the built image into the local Docker daemon when using docker/build-push-action v6 with push: false, set the load input to true. This corresponds to the Buildx --output=type=docker option, which exports the image to the Docker image store, making it available for commands like docker run. Example workflow step: - name: Build image uses: docker/build-push-action@v6 with: context: . push: false load: true tags: myimage:latest After this step, the image with the specified tags will be listed in docker images and usable locally. Note that load: true works for single-platform builds; for multi-platform, it loads only the current node platform's image. This behavior is consistent across v6 releases (up to v6.19.0 as of 2026-02) and is recommended in official Docker docs for testing before push workflows.

Citations:


🏁 Script executed:

cat .github/workflows/pyspectrafuse-containers.yml

Repository: bigbio/pyspectrafuse

Length of output: 4512


Add load: true to the docker/build-push-action configuration to enable local image testing on pull requests.

The workflow will fail the "Test container functionality" step on pull_request events because the image is never loaded into the local Docker daemon. The docker/build-push-action@v6 step sets push: false on PRs but lacks the required load: true parameter. With load: true, the image will be available locally for the subsequent docker run tests (the single-platform build with platforms: linux/amd64 supports this without issue).

Fix: Add load parameter
    - name: Build and Push pyspectrafuse image
      uses: docker/build-push-action@v6
      with:
        context: .
        push: ${{ (github.event.inputs.push_images == true || github.event.inputs.push_images == '') && (github.event_name != 'pull_request') }}
        load: ${{ github.event_name == 'pull_request' }}
        tags: |
          ghcr.io/bigbio/pyspectrafuse:latest
          ghcr.io/bigbio/pyspectrafuse:${{ steps.version.outputs.VERSION }}
        cache-from: type=gha
        cache-to: type=gha,mode=min
        build-args: |
          BUILDKIT_INLINE_CACHE=1
        provenance: false
        platforms: linux/amd64
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In @.github/workflows/pyspectrafuse-containers.yml around lines 79 - 85, The
build step using docker/build-push-action@v6 currently sets push: false for PRs
but doesn't set load, so the image isn't loaded into the local Docker daemon and
subsequent docker run tests fail; update the "Build and Push pyspectrafuse
image" action to add a load parameter (e.g., load: ${{ github.event_name ==
'pull_request' }} or load: true for PRs) so that when push is false on
pull_request the built image is loaded locally, ensuring the later docker run
tests (the commands invoking ghcr.io/bigbio/pyspectrafuse:${{
steps.version.outputs.VERSION }} and the Python import checks) can run against
the locally loaded image while keeping platforms: linux/amd64 and the existing
push conditional logic unchanged.

Comment on lines +94 to 96
uses: eWaterCycle/setup-singularity@v7
with:
singularity-version: 3.8.7

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🌐 Web query:

eWaterCycle/setup-singularity latest version 2026 Apptainer

💡 Result:

The latest version of the eWaterCycle/setup-singularity GitHub Action is v7, released on 2021-10-20. The repository explicitly recommends using the companion eWaterCycle/setup-apptainer action instead, as it supports newer versions of the open-source fork of Singularity (Apptainer). The setup-apptainer action's latest release is v2.0.0 (2022-10-25), with its last code push on 2025-01-07. For eWaterCycle workflows in 2026, official documentation prefers Apptainer over Singularity. Use uses: eWaterCycle/setup-apptainer@v2 with apptainer-version input (e.g., 1.3.6 or latest compatible). Example: - uses: eWaterCycle/setup-apptainer@v2 with: apptainer-version: 1.3.6 Apptainer is backward-compatible with Singularity images.

Citations:


Migrate from eWaterCycle/setup-singularity@v7 to eWaterCycle/setup-apptainer@v2.

The setup-singularity action (v7, released 2021-10-20) is outdated and explicitly deprecated by the maintainers. Since the Singularity/Apptainer split, the eWaterCycle repository recommends using setup-apptainer@v2 (last updated 2025-01-07) instead. Apptainer is the actively maintained fork of Singularity and is backward-compatible with Singularity images. Update to uses: eWaterCycle/setup-apptainer@v2 and use apptainer-version instead of singularity-version (e.g., apptainer-version: 1.3.6).

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In @.github/workflows/pyspectrafuse-containers.yml around lines 94 - 96, The
workflow uses the deprecated action uses: eWaterCycle/setup-singularity@v7 and
the singularity-version input; replace that with the maintained action uses:
eWaterCycle/setup-apptainer@v2 and swap the singularity-version key for
apptainer-version (for example apptainer-version: 1.3.6) so the job uses
Apptainer instead of the deprecated Singularity action.

…ipts

- Remove cluster_parquet_combine/ (CombineCluster2Parquet, ClusterResHandler),
  commands/cluster2parquet.py, common/sdrf_utils.py and their stale tests
  (test_sdrf_utils, test_integration_ftp). All cluster-TSV → parquet joins
  now go through scan_titles + DuckDB.
- Fold incremental merge-clusters into build-cluster-db via new
  --existing_metadata / --existing_membership flags. Only one CLI command
  needed for fresh and merging builds.
- incremental command group now only exposes extract-reps-dat.
- Rewrite incremental merge logic into a reusable merge_into_existing_db()
  in incremental/merge_results.py; extend resolve_clusters_dat to parse
  new-data titles and add load_new_spectra_with_arrays() DuckDB helper.
- Schema: add is_reused_cluster (bool) and source_datasets (list<string>)
  provenance columns; populate in both fresh and merge paths.
- spectrum2msp: honour empty instrument (collapses species/instrument/charge
  to species/charge for --skip_instrument runs).
- scripts/: add build_docker.sh and build_singularity.sh (local via
  docker-daemon or remote pull from ghcr.io; EBI Codon compatible SIF name).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
@ypriverol
ypriverol changed the base branch from main to dev April 18, 2026 07:15

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actionable comments posted: 8

♻️ Duplicate comments (6)
pyspectrafuse/commands/spectrum2msp.py (1)

356-357: ⚠️ Potential issue | 🟠 Major

Use the project directory for the MSP basename.

get_item_info() extracts from the last path segment, so passing the first parquet file can derive a filename-based basename instead of the project accession.

Proposed fix
-    basename = ParquetPathHandler(path_parquet_lst[0]).get_item_info()
+    basename = ParquetPathHandler(parquet_dir).get_item_info()
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/spectrum2msp.py` around lines 356 - 357, The basename
is derived by calling ParquetPathHandler(path_parquet_lst[0]).get_item_info(),
which uses the last path segment (file name) instead of the project accession;
change the code to derive the basename from the project directory rather than
the first parquet file so the MSP filenames use the project accession. Locate
the ParquetPathHandler usage and replace the input to get_item_info (or call the
appropriate helper on the project directory) so it receives the project
directory path (or its name) instead of path_parquet_lst[0], then keep building
output = output_dir / f"{basename}_{uuid.uuid4()}.msp.gz" as before.
README.md (1)

188-188: ⚠️ Potential issue | 🟡 Minor

Add a language identifier to the project-structure fence.

This still triggers markdownlint MD040; use text for the tree block.

Proposed fix
-```
+```text
 pyspectrafuse/
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@README.md` at line 188, The project-structure fence in README.md is missing a
language identifier and triggers markdownlint MD040; update the opening
triple-backtick for the project tree block (the fence containing
"pyspectrafuse/") to include the language identifier "text" so it reads as a
text code fence and resolves the lint error.
pyspectrafuse/common/duckdb_ops.py (2)

495-495: ⚠️ Potential issue | 🟡 Minor

Validate paired window inputs before zipping.

zip(window_tsvs, window_mz_ranges) silently truncates if the lists differ, producing incomplete merge results.

Proposed fix
-        for i, (tsv_path, (min_mz, max_mz)) in enumerate(zip(window_tsvs, window_mz_ranges)):
+        for i, (tsv_path, (min_mz, max_mz)) in enumerate(
+            zip(window_tsvs, window_mz_ranges, strict=True)
+        ):
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/common/duckdb_ops.py` at line 495, Before zipping window_tsvs
and window_mz_ranges, validate that their lengths match and fail fast: check
len(window_tsvs) == len(window_mz_ranges) (in the same function containing the
loop "for i, (tsv_path, (min_mz, max_mz)) in enumerate(zip(window_tsvs,
window_mz_ranges)):"), and raise a clear ValueError or RuntimeError if they
differ; alternatively, if partial pairing is intended, explicitly use
itertools.zip_longest with a clear handling strategy for missing entries. Ensure
the check occurs immediately before the zip/loop so mismatched inputs cannot
silently truncate results.

224-277: ⚠️ Potential issue | 🟠 Major

Avoid reading and writing the same membership parquet.

When output_path == existing_path, this COPY reads from and writes to the same file. That risks a torn/corrupt membership table if DuckDB truncates or the process fails mid-write. Write to a temp file in the same directory, then atomically replace.

Proposed fix
 def append_membership_dedup(existing_path: str,
                             new_df: pd.DataFrame,
                             output_path: str) -> str:
@@
-    conn = _create_connection()
+    final_output_path = output_path
+    existing_resolved = Path(existing_path).resolve()
+    output_resolved = Path(output_path).resolve()
+    if existing_resolved == output_resolved:
+        output_path = str(existing_resolved.with_suffix(
+            existing_resolved.suffix + ".tmp"))
+
+    conn = _create_connection()
@@
-        return output_path
+        if existing_resolved == output_resolved:
+            Path(output_path).replace(final_output_path)
+        return final_output_path
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/common/duckdb_ops.py` around lines 224 - 277, In
append_membership_dedup, avoid reading and writing to the same parquet by
detecting when output_path == existing_path and instead writing the COPY result
to a temporary file in the same directory (e.g., create a temp name next to
existing_path), then atomically replace the original with os.replace (or
equivalent) after the write completes; ensure the temp file is on the same
filesystem as existing_path and is cleaned up on errors, and keep all existing
logic that builds the SELECT and COPY but target the temp path when paths match.
pyspectrafuse/commands/build_cluster_db.py (1)

203-226: ⚠️ Potential issue | 🟠 Major

Bind SQL literals instead of interpolating user strings.

species and instrument are inserted directly into SQL; an apostrophe in either value breaks the COPY statement, and the same pattern applies to path literals.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/build_cluster_db.py` around lines 203 - 226, The SQL
currently built with an f-string in the conn.execute call interpolates user data
(species, instrument, membership_tmp, psm_path, source_union, charge_int) and
can break or allow injection when values contain quotes; stop using f-strings
and construct the query with proper SQL-safe building: use the psycopg/psycopg2
sql composition APIs (psycopg.sql.SQL with psycopg.sql.Literal for string/path
values and psycopg.sql.Identifier where appropriate) or use
conn.copy_expert/parameterized APIs so species and instrument (and file path
literals membership_tmp/psm_path and source_union/charge_int) are passed as
quoted Literals rather than interpolated strings; update the block around the
COPY (...) TO ... string in build_cluster_db.py that calls conn.execute to use
these safe constructors for each referenced symbol.
pyspectrafuse/incremental/merge_results.py (1)

242-248: ⚠️ Potential issue | 🟡 Minor

Coerce stats_purity.cluster_id before mapping stats.

meta['cluster_id'] is cast to str, but stats_purity['cluster_id'] keeps the parquet/DuckDB dtype. A mismatch silently maps no stats and leaves stale/default metadata.

Proposed fix
     meta['cluster_id'] = meta['cluster_id'].astype(str)
+    stats_purity['cluster_id'] = stats_purity['cluster_id'].astype(str)
 
     # Update stats from authoritative membership
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/incremental/merge_results.py` around lines 242 - 248, The
mapping silently fails because meta['cluster_id'] is cast to str but
stats_purity['cluster_id'] retains its original parquet/DuckDB dtype; coerce
stats_purity['cluster_id'] to str before creating stats_map or before set_index
to ensure keys match. Update the code around stats_map/stats_purity (e.g.,
convert stats_purity['cluster_id'] via astype(str) or use
stats_purity.assign(cluster_id=...).set_index('cluster_id')) so the subsequent
meta[col] = meta['cluster_id'].map(stats_map[col]).fillna(meta[col]) correctly
maps authoritative stats.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@pyspectrafuse/commands/build_cluster_db.py`:
- Around line 234-332: The code incorrectly labels single best-PSM spectra as
bin consensus when method_type == 'bin'; update the logic where chunk_meta is
constructed (and earlier where method_type is accepted) to prevent this: either
implement real binning (compute consensus_mz_array/consensus_intensity_array
from all member spectra via scan_parquet_arrays or a new binning function) or
disallow/override 'bin' for fresh builds by raising an error or forcing
method_type='best' and setting consensus_method='best' so you don't write
consensus_method='bin' while using best_keys/scan_parquet_arrays; modify
references to method_type, consensus_method, best_keys, chunk_meta, and
scan_parquet_arrays accordingly.
- Around line 133-196: The code loses dataset provenance for each parquet
(all_parquet_files) so the DuckDB source UNION (built in source_parts) and
subsequent LEFT JOINs match only on (run_file_name, orig_scan) and can
cross-enrich rows from different datasets; to fix, preserve and propagate
dataset identity: change how all_parquet_files is built (return or store tuples
like (ppath, dataset_name) or derive dataset from the parquet_dirs loop), add a
literal column in each source_parts SELECT (e.g. 'my_dataset' AS dataset_name)
alongside run_file_name/orig_scan/precursor_mz etc., and then update the JOIN
logic that enriches membership to include dataset_name in the join key (use
dataset_name plus run_file_name and orig_scan) so enrichment and deduplication
remain dataset-scoped; adjust any downstream references to include dataset_name
and update find_target_ext_files/_detect_scan_expr usage accordingly.
- Around line 398-431: The current strict check len(scan_titles) ==
len(dataset_name) prevents incremental merging; change the validation so that
when existing_metadata and existing_membership are provided we do not require
scan_titles and dataset_name to be one-to-one — instead require at least one
scan_title and at least one dataset_name and disallow multiple dataset_name
entries for incremental mode (so project_accession can be dataset_name[0]); keep
the original one-to-one check for the non-incremental path. Update the
conditional around the validation and ensure merge_into_existing_db(...) still
receives project_accession=dataset_name[0] and
scan_titles_files=list(scan_titles).

In `@pyspectrafuse/commands/spectrum2msp.py`:
- Around line 211-223: The SELECT that is appended into parts (the SQL block
using run_col, scan_expr, mz_array, intensity_array, charge_col, mz_expr,
pep_col, qvalue_expr from read_parquet('{ppath}')) must also project a dataset
identifier so rows can be joined by (dataset, run_file_name, orig_scan); modify
that SELECT to include a literal or column that identifies the source dataset
(e.g., AS dataset) and ensure subsequent JOINs use membership.m.dataset =
source.dataset in addition to run_file_name/orig_scan; apply the same change to
the other UNIONed block at the later location (lines 264–270) so both UNION arms
are tagged and joined by dataset.

In `@pyspectrafuse/common/duckdb_ops.py`:
- Around line 471-527: The current de-dup logic in merge_mz_window_clusters
always picks the lowest window_idx and ignores
overlap_da/window_min_mz/window_max_mz; change it so the dedupe prefers the
non-overlap ("home") assignment using overlap metadata: either require the input
TSVs to include the spectrum m/z (add a 'spectrum_mz' column when reading CSVs)
or update the function contract to accept per-spectrum mz, then compute a
boolean in_overlap per row (e.g., in_overlap = spectrum_mz >= window_min_mz and
spectrum_mz <= window_max_mz and within the overlap bounds computed from
overlap_da), register that boolean into combined, and modify the SQL dedupe
ORDER BY (in the query against the registered table 'combined' used by
merge_mz_window_clusters) to ORDER BY in_overlap ASC, window_idx ASC so
non-overlap rows are preferred; ensure symbols referenced:
merge_mz_window_clusters, overlap_da, window_min_mz, window_max_mz, combined,
and window_idx.

In `@pyspectrafuse/incremental/merge_results.py`:
- Around line 203-205: Groupby.idxmin can fail when all 'global_qvalue' values
for a cluster are NaN; before computing best_new_idx use a deterministic fill
(e.g. replace NaN with np.inf) on the 'global_qvalue' column of
new_for_existing_df so groupby('cluster_accession')['global_qvalue'].idxmin()
always returns a valid index; apply the same fix to the analogous block around
lines 320–326 and ensure numpy (np) is imported/available where you make the
replacement.
- Around line 108-110: The current branch that checks new_df.empty logs a
warning but still returns out_metadata and out_membership (which point to file
paths that were never written); change the empty-DF early-return in
merge_results.py so it does not return non-existent file paths—e.g., after
logger.warning("No new spectra matched cluster assignments") return (None, None)
or another explicit sentinel instead of out_metadata/out_membership so
callers/CLI can detect no files were written; update any callers that unpack the
return to handle the sentinel accordingly (refer to new_df.empty,
logger.warning, and the existing return out_metadata, out_membership).

In `@README.md`:
- Around line 140-151: The README example uses the removed "pyspectrafuse
incremental merge-clusters" command but incremental.py now only registers
"extract-reps-dat" and merging is performed by the "build-cluster-db" command;
update the documentation to replace the deprecated "incremental merge-clusters"
invocation with the equivalent "build-cluster-db" usage including the existing
metadata and membership flags (--existing_metadata and --existing_membership)
and preserve the other flags shown (--new_parquet_dir, --species, --instrument,
--charge, --method_type) so users are pointed at the correct CLI entrypoint.

---

Duplicate comments:
In `@pyspectrafuse/commands/build_cluster_db.py`:
- Around line 203-226: The SQL currently built with an f-string in the
conn.execute call interpolates user data (species, instrument, membership_tmp,
psm_path, source_union, charge_int) and can break or allow injection when values
contain quotes; stop using f-strings and construct the query with proper
SQL-safe building: use the psycopg/psycopg2 sql composition APIs
(psycopg.sql.SQL with psycopg.sql.Literal for string/path values and
psycopg.sql.Identifier where appropriate) or use conn.copy_expert/parameterized
APIs so species and instrument (and file path literals membership_tmp/psm_path
and source_union/charge_int) are passed as quoted Literals rather than
interpolated strings; update the block around the COPY (...) TO ... string in
build_cluster_db.py that calls conn.execute to use these safe constructors for
each referenced symbol.

In `@pyspectrafuse/commands/spectrum2msp.py`:
- Around line 356-357: The basename is derived by calling
ParquetPathHandler(path_parquet_lst[0]).get_item_info(), which uses the last
path segment (file name) instead of the project accession; change the code to
derive the basename from the project directory rather than the first parquet
file so the MSP filenames use the project accession. Locate the
ParquetPathHandler usage and replace the input to get_item_info (or call the
appropriate helper on the project directory) so it receives the project
directory path (or its name) instead of path_parquet_lst[0], then keep building
output = output_dir / f"{basename}_{uuid.uuid4()}.msp.gz" as before.

In `@pyspectrafuse/common/duckdb_ops.py`:
- Line 495: Before zipping window_tsvs and window_mz_ranges, validate that their
lengths match and fail fast: check len(window_tsvs) == len(window_mz_ranges) (in
the same function containing the loop "for i, (tsv_path, (min_mz, max_mz)) in
enumerate(zip(window_tsvs, window_mz_ranges)):"), and raise a clear ValueError
or RuntimeError if they differ; alternatively, if partial pairing is intended,
explicitly use itertools.zip_longest with a clear handling strategy for missing
entries. Ensure the check occurs immediately before the zip/loop so mismatched
inputs cannot silently truncate results.
- Around line 224-277: In append_membership_dedup, avoid reading and writing to
the same parquet by detecting when output_path == existing_path and instead
writing the COPY result to a temporary file in the same directory (e.g., create
a temp name next to existing_path), then atomically replace the original with
os.replace (or equivalent) after the write completes; ensure the temp file is on
the same filesystem as existing_path and is cleaned up on errors, and keep all
existing logic that builds the SELECT and COPY but target the temp path when
paths match.

In `@pyspectrafuse/incremental/merge_results.py`:
- Around line 242-248: The mapping silently fails because meta['cluster_id'] is
cast to str but stats_purity['cluster_id'] retains its original parquet/DuckDB
dtype; coerce stats_purity['cluster_id'] to str before creating stats_map or
before set_index to ensure keys match. Update the code around
stats_map/stats_purity (e.g., convert stats_purity['cluster_id'] via astype(str)
or use stats_purity.assign(cluster_id=...).set_index('cluster_id')) so the
subsequent meta[col] = meta['cluster_id'].map(stats_map[col]).fillna(meta[col])
correctly maps authoritative stats.

In `@README.md`:
- Line 188: The project-structure fence in README.md is missing a language
identifier and triggers markdownlint MD040; update the opening triple-backtick
for the project tree block (the fence containing "pyspectrafuse/") to include
the language identifier "text" so it reads as a text code fence and resolves the
lint error.
🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

  • Push a commit to this branch (recommended)
  • Create a new PR with the fixes

ℹ️ Review info
⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 87fa48f2-6ee1-4945-9f3e-75b182f6e9c6

📥 Commits

Reviewing files that changed from the base of the PR and between 4628df8 and d74e4b6.

📒 Files selected for processing (16)
  • README.md
  • pyspectrafuse/cluster_parquet_combine/cluster_res_handler.py
  • pyspectrafuse/cluster_parquet_combine/combine_cluster_and_parquet.py
  • pyspectrafuse/commands/build_cluster_db.py
  • pyspectrafuse/commands/incremental.py
  • pyspectrafuse/commands/spectrum2msp.py
  • pyspectrafuse/common/duckdb_ops.py
  • pyspectrafuse/common/schemas.py
  • pyspectrafuse/common/sdrf_utils.py
  • pyspectrafuse/incremental/merge_results.py
  • pyspectrafuse/incremental/resolve_clusters_dat.py
  • pyspectrafuse/pyspectrafuse.py
  • scripts/build_docker.sh
  • scripts/build_singularity.sh
  • tests/test_integration_ftp.py
  • tests/test_sdrf_utils.py
💤 Files with no reviewable changes (5)
  • tests/test_sdrf_utils.py
  • pyspectrafuse/common/sdrf_utils.py
  • pyspectrafuse/cluster_parquet_combine/cluster_res_handler.py
  • tests/test_integration_ftp.py
  • pyspectrafuse/cluster_parquet_combine/combine_cluster_and_parquet.py
✅ Files skipped from review due to trivial changes (2)
  • pyspectrafuse/common/schemas.py
  • scripts/build_docker.sh
🚧 Files skipped from review as they are similar to previous changes (1)
  • pyspectrafuse/pyspectrafuse.py

Comment on lines +133 to +196
# Collect all parquet file paths
all_parquet_files = []
for pdir in parquet_dirs:
all_parquet_files.extend(find_target_ext_files(pdir, '.parquet'))

# ── Pass 1: Build PSM membership entirely in DuckDB ──
# DuckDB reads source parquet with predicate pushdown (charge filter)
# and SEMI JOINs with the membership temp parquet — no pandas intermediary.
logger.info(f"Pass 1: Building PSM membership via DuckDB (charge={charge_int})...")

# Use a file-backed DuckDB database so large JOINs can spill to disk
# instead of OOM-killing the process.
duckdb_path = str(Path(tmp_dir) / 'build.duckdb')
conn = duckdb.connect(duckdb_path)
try:
conn.execute("SET memory_limit='4GB'")
conn.execute(f"SET temp_directory='{tmp_dir}'")
conn.execute("SET threads=4")
except Exception:
conn.close()
raise

# Build the source parquet UNION with column normalization
source_parts = []
for ppath in all_parquet_files:
cols_df = conn.execute(f"SELECT name FROM parquet_schema('{ppath}')").fetchdf()
available = set(cols_df['name'])

run_col = 'run_file_name' if 'run_file_name' in available else 'reference_file_name'
charge_col = 'charge' if 'charge' in available else 'precursor_charge'

mz_parts = []
if 'observed_mz' in available: mz_parts.append('observed_mz')
if 'calculated_mz' in available: mz_parts.append('calculated_mz')
if 'exp_mass_to_charge' in available: mz_parts.append('exp_mass_to_charge')
mz_expr = f"COALESCE({', '.join(mz_parts)}, 0.0)" if mz_parts else '0.0'

pep_col = 'posterior_error_probability' if 'posterior_error_probability' in available else 'NULL'

# Handle scan column type
scan_expr = _detect_scan_expr(conn, ppath)

if 'global_qvalue' in available:
qvalue_expr = 'global_qvalue'
elif 'additional_scores' in available:
qvalue_expr = """(
SELECT s.score_value
FROM UNNEST(additional_scores) AS t(s)
WHERE s.score_name = 'global_qvalue'
LIMIT 1
)"""
else:
qvalue_expr = 'NULL'

source_parts.append(f"""
SELECT
{run_col} AS run_file_name,
{scan_expr} AS orig_scan,
{mz_expr} AS precursor_mz,
{pep_col} AS posterior_error_probability,
{qvalue_expr} AS global_qvalue
FROM read_parquet('{ppath}')
WHERE {charge_col} = {charge_int}
""")

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Preserve dataset ownership when enriching membership.

all_parquet_files loses which --dataset_name each parquet came from, then the LEFT JOIN matches only (run_file_name, orig_scan). If two datasets contain the same run filename and scan number, rows can be enriched from the wrong parquet or duplicated.

Suggested direction
-    all_parquet_files = []
-    for pdir in parquet_dirs:
-        all_parquet_files.extend(find_target_ext_files(pdir, '.parquet'))
+    all_parquet_files = []
+    for pdir, ds_name in zip(parquet_dirs, dataset_names, strict=True):
+        for ppath in find_target_ext_files(pdir, '.parquet'):
+            all_parquet_files.append((ppath, ds_name))
...
-    for ppath in all_parquet_files:
+    for ppath, ds_name in all_parquet_files:
...
             SELECT
+                '{ds_name}' AS dataset,
                 {run_col} AS run_file_name,
...
-                    AND m.orig_scan = s.orig_scan
+                    AND m.orig_scan = s.orig_scan
+                    AND m.dataset = s.dataset

Also applies to: 203-223

🧰 Tools
🪛 Ruff (0.15.10)

[error] 158-158: Possible SQL injection vector through string-based query construction

(S608)


[error] 165-165: Multiple statements on one line (colon)

(E701)


[error] 166-166: Multiple statements on one line (colon)

(E701)


[error] 167-167: Multiple statements on one line (colon)

(E701)


[error] 187-196: Possible SQL injection vector through string-based query construction

(S608)

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/build_cluster_db.py` around lines 133 - 196, The code
loses dataset provenance for each parquet (all_parquet_files) so the DuckDB
source UNION (built in source_parts) and subsequent LEFT JOINs match only on
(run_file_name, orig_scan) and can cross-enrich rows from different datasets; to
fix, preserve and propagate dataset identity: change how all_parquet_files is
built (return or store tuples like (ppath, dataset_name) or derive dataset from
the parquet_dirs loop), add a literal column in each source_parts SELECT (e.g.
'my_dataset' AS dataset_name) alongside run_file_name/orig_scan/precursor_mz
etc., and then update the JOIN logic that enriches membership to include
dataset_name in the join key (use dataset_name plus run_file_name and orig_scan)
so enrichment and deduplication remain dataset-scoped; adjust any downstream
references to include dataset_name and update
find_target_ext_files/_detect_scan_expr usage accordingly.

Comment on lines +234 to +332
# ── Best PSM per cluster + spectrum arrays (chunked to avoid OOM) ──
logger.info(f"Pass 2: Finding best PSM per cluster...")

# Get best PSM keys via DuckDB on the membership parquet we just wrote
best_keys = conn.execute(f"""
SELECT cluster_id, reference_file_name AS run_file_name,
scan AS orig_scan, peptidoform, precursor_mz
FROM (
SELECT *,
ROW_NUMBER() OVER (
PARTITION BY cluster_id
ORDER BY posterior_error_probability ASC NULLS LAST
) AS _rn
FROM read_parquet('{psm_path}')
)
WHERE _rn = 1
""").fetchdf()
n_clusters = len(best_keys)
best_keys['orig_scan'] = best_keys['orig_scan'].astype(int)
logger.info(f"Best PSMs: {n_clusters} clusters")
finally:
conn.close()

# Merge with stats/purity early (small — just scalars)
best_keys = best_keys.merge(stats_purity, on='cluster_id', how='left')
best_keys['purity'] = best_keys['purity'].fillna(1.0).astype(np.float32)
best_keys['member_count'] = best_keys['member_count'].fillna(1).astype(np.int32)
best_keys['project_count'] = best_keys['project_count'].fillna(1).astype(np.int16)
del stats_purity; gc.collect()

# Aggregate source_datasets per cluster (DISTINCT project_accession list)
conn_prov = duckdb.connect(':memory:')
try:
sources_df = conn_prov.execute(f"""
SELECT cluster_id,
list_distinct(array_agg(project_accession)) AS source_datasets
FROM read_parquet('{psm_path}')
GROUP BY cluster_id
""").fetchdf()
finally:
conn_prov.close()
best_keys = best_keys.merge(sources_df, on='cluster_id', how='left')
best_keys['source_datasets'] = best_keys['source_datasets'].apply(
lambda x: list(x) if isinstance(x, (list, np.ndarray)) else list(set(dataset_names)))
del sources_df; gc.collect()

# ── Write cluster metadata in chunks (avoids loading all arrays at once) ──
meta_path = str(out_dir / 'cluster_metadata.parquet')
writer = None
total_written = 0

def to_f32_list(arr):
if arr is None:
return []
if isinstance(arr, np.ndarray):
return arr.astype(np.float32).tolist()
if isinstance(arr, (list, tuple)):
return [float(x) for x in arr]
return []

for chunk_start in range(0, n_clusters, chunk_size):
chunk_end = min(chunk_start + chunk_size, n_clusters)
chunk_keys = best_keys.iloc[chunk_start:chunk_end]
logger.info(f"Pass 2 chunk {chunk_start//chunk_size + 1}: "
f"clusters {chunk_start}-{chunk_end} of {n_clusters}")

# Read arrays for this chunk only
keys_for_join = chunk_keys[['run_file_name', 'orig_scan']].copy()
arrays_df = scan_parquet_arrays(
all_parquet_files, keys_for_join, charge_int,
temp_directory=tmp_dir)

chunk_with_arrays = chunk_keys.merge(
arrays_df, on=['run_file_name', 'orig_scan'], how='left')
del arrays_df, keys_for_join; gc.collect()

has_mz = 'mz_array' in chunk_with_arrays.columns
has_int = 'intensity_array' in chunk_with_arrays.columns

chunk_meta = pd.DataFrame({
'cluster_id': chunk_with_arrays['cluster_id'].values,
'species': species,
'instrument': instrument,
'charge': np.int8(charge_int),
'peptidoform': chunk_with_arrays['peptidoform'].values,
'peptide_sequence': chunk_with_arrays['peptidoform'].str.split('/').str[0].str.replace(
r'\[.*?\]', '', regex=True).values,
'consensus_mz_array': chunk_with_arrays['mz_array'].apply(to_f32_list).values if has_mz else [[] for _ in range(len(chunk_with_arrays))],
'consensus_intensity_array': chunk_with_arrays['intensity_array'].apply(to_f32_list).values if has_int else [[] for _ in range(len(chunk_with_arrays))],
'consensus_method': method_type,
'precursor_mz': pd.to_numeric(chunk_with_arrays['precursor_mz'], errors='coerce').fillna(0).values,
'member_count': chunk_with_arrays['member_count'].values,
'project_count': chunk_with_arrays['project_count'].values,
'best_pep': chunk_with_arrays['best_pep'].values if 'best_pep' in chunk_with_arrays.columns else np.nan,
'best_qvalue': chunk_with_arrays['best_qvalue'].values if 'best_qvalue' in chunk_with_arrays.columns else np.nan,
'purity': chunk_with_arrays['purity'].values,
'is_reused_cluster': False,
'source_datasets': chunk_with_arrays['source_datasets'].values if 'source_datasets' in chunk_with_arrays.columns else [list(set(dataset_names)) for _ in range(len(chunk_with_arrays))],
})

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Do not label best-PSM spectra as bin consensus.

Fresh builds accept --method_type bin, but this path still selects one best PSM per cluster and writes its arrays directly. That produces metadata marked consensus_method='bin' without running the binning strategy.

Either compute bin consensus from all member spectra for each cluster, or restrict fresh build-cluster-db to best until binning is implemented here.

🧰 Tools
🪛 Ruff (0.15.10)

[error] 235-235: f-string without any placeholders

Remove extraneous f prefix

(F541)


[error] 238-250: Possible SQL injection vector through string-based query construction

(S608)


[error] 262-262: Multiple statements on one line (semicolon)

(E702)


[error] 267-272: Possible SQL injection vector through string-based query construction

(S608)


[error] 278-278: Multiple statements on one line (semicolon)

(E702)


[error] 308-308: Multiple statements on one line (semicolon)

(E702)

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/build_cluster_db.py` around lines 234 - 332, The code
incorrectly labels single best-PSM spectra as bin consensus when method_type ==
'bin'; update the logic where chunk_meta is constructed (and earlier where
method_type is accepted) to prevent this: either implement real binning (compute
consensus_mz_array/consensus_intensity_array from all member spectra via
scan_parquet_arrays or a new binning function) or disallow/override 'bin' for
fresh builds by raising an error or forcing method_type='best' and setting
consensus_method='best' so you don't write consensus_method='bin' while using
best_keys/scan_parquet_arrays; modify references to method_type,
consensus_method, best_keys, chunk_meta, and scan_parquet_arrays accordingly.

Comment on lines +398 to +431
if len(scan_titles) != len(dataset_name):
raise click.ClickException(
f"Number of --scan_titles ({len(scan_titles)}) must match "
f"--dataset_name ({len(dataset_name)})")
if bool(existing_metadata) != bool(existing_membership):
raise click.ClickException(
"--existing_metadata and --existing_membership must be provided together")

if existing_metadata and existing_membership:
from pyspectrafuse.incremental.merge_results import merge_into_existing_db
# Merge into existing DB. Uses the FIRST parquet_dir/dataset_name as the
# "new data" source — callers invoking this mode are expected to pass
# exactly one new-data parquet_dir plus any number of rep scan_titles.
from pyspectrafuse.commands.spectrum2msp import find_target_ext_files
new_parquet_paths = []
for pdir in parquet_dir:
new_parquet_paths.extend(find_target_ext_files(pdir, '.parquet'))
if not new_parquet_paths:
raise click.ClickException(
f"No parquet files found in provided --parquet_dir entries")

meta_path, psm_path = merge_into_existing_db(
cluster_tsv=cluster_tsv,
scan_titles_files=list(scan_titles),
existing_metadata_path=existing_metadata,
existing_membership_path=existing_membership,
new_parquet_paths=new_parquet_paths,
species=species,
instrument=instrument,
charge=charge,
method_type=method_type,
output_dir=output_dir,
project_accession=dataset_name[0],
)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Relax incremental-mode scan-title validation.

Incremental merging needs rep scan-title files plus new-data scan-title files, but only the new dataset needs project_accession. The unconditional len(scan_titles) == len(dataset_name) check rejects the documented workflow unless callers invent dataset names for rep files, and project_accession=dataset_name[0] can then point at the wrong item.

Suggested direction
-    if len(scan_titles) != len(dataset_name):
+    if not (existing_metadata and existing_membership) and len(scan_titles) != len(dataset_name):
         raise click.ClickException(
             f"Number of --scan_titles ({len(scan_titles)}) must match "
             f"--dataset_name ({len(dataset_name)})")
...
     if existing_metadata and existing_membership:
+        if len(dataset_name) != 1:
+            raise click.ClickException(
+                "Incremental merge expects exactly one --dataset_name for the new data")
🧰 Tools
🪛 Ruff (0.15.10)

[error] 417-417: f-string without any placeholders

Remove extraneous f prefix

(F541)

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/build_cluster_db.py` around lines 398 - 431, The
current strict check len(scan_titles) == len(dataset_name) prevents incremental
merging; change the validation so that when existing_metadata and
existing_membership are provided we do not require scan_titles and dataset_name
to be one-to-one — instead require at least one scan_title and at least one
dataset_name and disallow multiple dataset_name entries for incremental mode (so
project_accession can be dataset_name[0]); keep the original one-to-one check
for the non-incremental path. Update the conditional around the validation and
ensure merge_into_existing_db(...) still receives
project_accession=dataset_name[0] and scan_titles_files=list(scan_titles).

Comment on lines +211 to +223
parts.append(f"""
SELECT
{run_col} AS run_file_name,
{scan_expr} AS orig_scan,
mz_array,
intensity_array,
CAST({charge_col} AS INTEGER) AS charge,
{mz_expr} AS pepmass,
{pep_col} AS posterior_error_probability,
{qvalue_expr} AS global_qvalue
FROM read_parquet('{ppath}')
WHERE {charge_col} = {charge_int}
""")

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Include dataset identity in the DuckDB join.

For multi-dataset MSP generation, (run_file_name, orig_scan) is not globally unique. Since membership already carries m.dataset, the source UNION should also tag each parquet row with its dataset and join on that key to avoid pulling spectra from the wrong project.

Suggested direction
-                {run_col} AS run_file_name,
+                {dataset_expr} AS dataset,
+                {run_col} AS run_file_name,
...
-            ON s.run_file_name = m.run_file_name
+            ON s.dataset = m.dataset
+            AND s.run_file_name = m.run_file_name
             AND s.orig_scan = m.orig_scan

Also applies to: 264-270

🧰 Tools
🪛 Ruff (0.15.10)

[error] 211-223: Possible SQL injection vector through string-based query construction

(S608)

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/commands/spectrum2msp.py` around lines 211 - 223, The SELECT
that is appended into parts (the SQL block using run_col, scan_expr, mz_array,
intensity_array, charge_col, mz_expr, pep_col, qvalue_expr from
read_parquet('{ppath}')) must also project a dataset identifier so rows can be
joined by (dataset, run_file_name, orig_scan); modify that SELECT to include a
literal or column that identifies the source dataset (e.g., AS dataset) and
ensure subsequent JOINs use membership.m.dataset = source.dataset in addition to
run_file_name/orig_scan; apply the same change to the other UNIONed block at the
later location (lines 264–270) so both UNION arms are tagged and joined by
dataset.

Comment on lines +471 to +527
def merge_mz_window_clusters(
window_tsvs: List[str],
window_mz_ranges: List[tuple],
overlap_da: float = 1.0,
) -> pd.DataFrame:
"""Merge cluster assignments from overlapping m/z windows.

Spectra in overlap zones may be assigned to clusters in both windows.
Resolution: if a spectrum is in an overlap zone and was clustered in both
windows, prefer the assignment from the window where it is NOT in the
overlap zone (i.e., from its "home" window). If in overlap in both, keep
the one from the lower window.

Args:
window_tsvs: List of MaRaCluster output TSV paths, one per window.
window_mz_ranges: List of (min_mz, max_mz) tuples for each window.
overlap_da: Overlap width in Da.

Returns:
DataFrame with columns: mgf_path, scannr, cluster_id (resolved).
"""
conn = _create_connection()
try:
all_parts = []
for i, (tsv_path, (min_mz, max_mz)) in enumerate(zip(window_tsvs, window_mz_ranges)):
df = pd.read_csv(tsv_path, sep='\t', header=None,
names=['mgf_path', 'scannr', 'cluster_id'],
skip_blank_lines=False)
df = df.dropna()
df['window_idx'] = i
df['window_min_mz'] = min_mz
df['window_max_mz'] = max_mz
# Prefix cluster IDs with window index to avoid collisions
df['cluster_id'] = f'w{i}_' + df['cluster_id'].astype(str)
all_parts.append(df)

if not all_parts:
return pd.DataFrame()

combined = pd.concat(all_parts, ignore_index=True)
conn.register('combined', combined)

# For spectra appearing in multiple windows (overlap zone), keep the
# assignment from the window with the lower index (deterministic).
# Dedup by scannr only — within a charge partition, scannr is unique.
# mgf_path differs across windows for the same spectrum.
result = conn.execute("""
SELECT mgf_path, scannr, cluster_id FROM (
SELECT *,
ROW_NUMBER() OVER (
PARTITION BY scannr
ORDER BY window_idx
) AS _rn
FROM combined
)
WHERE _rn = 1
""").fetchdf()

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Use the overlap window metadata or update the contract.

The docstring says overlap assignments prefer the non-overlap “home” window, but the query always keeps the lowest window_idx; overlap_da, window_min_mz, and window_max_mz never affect resolution. This can select the wrong cluster assignment in overlapping windows.

🧰 Tools
🪛 Ruff (0.15.10)

[warning] 495-495: zip() without an explicit strict= parameter

Add explicit value for parameter strict=

(B905)

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/common/duckdb_ops.py` around lines 471 - 527, The current
de-dup logic in merge_mz_window_clusters always picks the lowest window_idx and
ignores overlap_da/window_min_mz/window_max_mz; change it so the dedupe prefers
the non-overlap ("home") assignment using overlap metadata: either require the
input TSVs to include the spectrum m/z (add a 'spectrum_mz' column when reading
CSVs) or update the function contract to accept per-spectrum mz, then compute a
boolean in_overlap per row (e.g., in_overlap = spectrum_mz >= window_min_mz and
spectrum_mz <= window_max_mz and within the overlap bounds computed from
overlap_da), register that boolean into combined, and modify the SQL dedupe
ORDER BY (in the query against the registered table 'combined' used by
merge_mz_window_clusters) to ORDER BY in_overlap ASC, window_idx ASC so
non-overlap rows are preferred; ensure symbols referenced:
merge_mz_window_clusters, overlap_da, window_min_mz, window_max_mz, combined,
and window_idx.

Comment on lines +108 to +110
if new_df.empty:
logger.warning("No new spectra matched cluster assignments")
return out_metadata, out_membership

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Do not return paths for files that were not written.

When no new spectra match, this returns out_metadata / out_membership even though neither file is created. The CLI then reports success paths that may not exist.

Proposed fix
     if new_df.empty:
         logger.warning("No new spectra matched cluster assignments")
-        return out_metadata, out_membership
+        return None, None
📝 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.

Suggested change
if new_df.empty:
logger.warning("No new spectra matched cluster assignments")
return out_metadata, out_membership
if new_df.empty:
logger.warning("No new spectra matched cluster assignments")
return None, None
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/incremental/merge_results.py` around lines 108 - 110, The
current branch that checks new_df.empty logs a warning but still returns
out_metadata and out_membership (which point to file paths that were never
written); change the empty-DF early-return in merge_results.py so it does not
return non-existent file paths—e.g., after logger.warning("No new spectra
matched cluster assignments") return (None, None) or another explicit sentinel
instead of out_metadata/out_membership so callers/CLI can detect no files were
written; update any callers that unpack the return to handle the sentinel
accordingly (refer to new_df.empty, logger.warning, and the existing return
out_metadata, out_membership).

Comment on lines +203 to +205
# BEST: swap spectrum when new PSM has better qvalue
best_new_idx = new_for_existing_df.groupby('cluster_accession')['global_qvalue'].idxmin()
best_new = new_for_existing_df.loc[best_new_idx].set_index('cluster_accession')

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Guard idxmin() against all-null q-values.

If a cluster has only missing global_qvalue, groupby().idxmin() can produce invalid indexers. Fill with np.inf before grouping so a deterministic row is still selected.

Proposed fix
-            best_new_idx = new_for_existing_df.groupby('cluster_accession')['global_qvalue'].idxmin()
+            ranked_new = new_for_existing_df.copy()
+            ranked_new['_global_qvalue'] = pd.to_numeric(
+                ranked_new['global_qvalue'], errors='coerce'
+            ).fillna(np.inf)
+            best_new_idx = ranked_new.groupby('cluster_accession')['_global_qvalue'].idxmin()
-            best_new = new_for_existing_df.loc[best_new_idx].set_index('cluster_accession')
+            best_new = ranked_new.loc[best_new_idx].set_index('cluster_accession')
...
-    best_idx = df.groupby('cluster_accession')['global_qvalue'].idxmin()
-    best = df.loc[best_idx].set_index('cluster_accession')
+    ranked = df.copy()
+    ranked['_global_qvalue'] = pd.to_numeric(
+        ranked['global_qvalue'], errors='coerce'
+    ).fillna(np.inf)
+    best_idx = ranked.groupby('cluster_accession')['_global_qvalue'].idxmin()
+    best = ranked.loc[best_idx].set_index('cluster_accession')

Also applies to: 320-326

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@pyspectrafuse/incremental/merge_results.py` around lines 203 - 205,
Groupby.idxmin can fail when all 'global_qvalue' values for a cluster are NaN;
before computing best_new_idx use a deterministic fill (e.g. replace NaN with
np.inf) on the 'global_qvalue' column of new_for_existing_df so
groupby('cluster_accession')['global_qvalue'].idxmin() always returns a valid
index; apply the same fix to the analogous block around lines 320–326 and ensure
numpy (np) is imported/available where you make the replacement.

Comment thread README.md
Comment on lines +140 to +151
# Step 2: After MaRaCluster runs on [representatives + new data], merge results
pyspectrafuse incremental merge-clusters \
--cluster_tsv new_clusters.tsv \
--scan_titles_dir scan_titles/ \
--existing_metadata cluster_db/cluster_metadata.parquet \
--existing_membership cluster_db/psm_cluster_membership.parquet \
--new_parquet_dir /path/to/new_project/ \
--species "Homo sapiens" \
--instrument "Q Exactive HF" \
--charge charge2 \
--method_type bin
```

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Replace the removed incremental merge-clusters command in the docs.

incremental.py only registers extract-reps-dat; merge is now handled by build-cluster-db with --existing_metadata and --existing_membership. As written, users following this example will get a missing-command error.

Proposed documentation fix
-# Step 2: After MaRaCluster runs on [representatives + new data], merge results
-pyspectrafuse incremental merge-clusters \
+# Step 2: After MaRaCluster runs on [representatives + new data], merge results
+pyspectrafuse build-cluster-db \
     --cluster_tsv new_clusters.tsv \
-    --scan_titles_dir scan_titles/ \
+    --scan_titles reps_output/reps_cluster_metadata.scan_titles.txt \
+    --scan_titles dat_output/new_project.psm_charge2.scan_titles.txt \
+    --parquet_dir /path/to/new_project/ \
+    --dataset_name PXD_NEW \
     --existing_metadata cluster_db/cluster_metadata.parquet \
     --existing_membership cluster_db/psm_cluster_membership.parquet \
-    --new_parquet_dir /path/to/new_project/ \
     --species "Homo sapiens" \
     --instrument "Q Exactive HF" \
     --charge charge2 \
-    --method_type bin
+    --method_type bin \
+    --output_dir merged_cluster_db/
...
-│   ├── incremental.py              # incremental sub-commands (extract-reps-dat, merge-clusters)
+│   ├── incremental.py              # incremental sub-commands (extract-reps-dat)

Also applies to: 197-197

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@README.md` around lines 140 - 151, The README example uses the removed
"pyspectrafuse incremental merge-clusters" command but incremental.py now only
registers "extract-reps-dat" and merging is performed by the "build-cluster-db"
command; update the documentation to replace the deprecated "incremental
merge-clusters" invocation with the equivalent "build-cluster-db" usage
including the existing metadata and membership flags (--existing_metadata and
--existing_membership) and preserve the other flags shown (--new_parquet_dir,
--species, --instrument, --charge, --method_type) so users are pointed at the
correct CLI entrypoint.

ypriverol and others added 3 commits April 18, 2026 08:18
- test_incremental_help: assert `merge-clusters` is NOT advertised
  (the sub-command was folded into `build-cluster-db`).
- Drop the now-invalid test_incremental_merge_clusters_help.
- Add test_build_cluster_db_help_mentions_merge_flags: asserts
  --existing_metadata / --existing_membership appear in build-cluster-db
  help.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Add conditional loading for pull request events.
Tuple-of-rows DataFrame construction coerces Python None → NaN on newer
pandas (seen on CI: 3.11/3.12 with pandas 2.x) when the row mixes None
with ints/strs across columns. The test asserts
`df.iloc[2]['old_cluster_id'] is None`, which fails when the value is NaN.

Rebuild the DataFrame column-wise and explicitly restore None in the
three string columns that can legitimately be missing
(old_cluster_id, run_file_name, peptidoform).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
@ypriverol
ypriverol merged commit 979a791 into dev Apr 18, 2026
7 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant