Claude/review sample brain 01 c mi2nn8yi8 f2 e bvcd6 rcoc

[jannekbuengener]

Summary by Sourcery

Expand the sample management pipeline to a DAW-neutral, multi-format export framework with streaming support and dedicated adapters for seven major DAWs; introduce a comprehensive metadata consolidation module and an optional EDM-optimized analysis mode with new database schema extensions; refactor existing FL export logic and enhance the CLI with unified export commands and view creation.

New Features:

• Add DAW-neutral export command supporting JSON, CSV, YAML, XML, and Parquet formats with optional streaming for large libraries
• Implement DAW-specific export adapters for Ableton, Bitwig, FL Studio, Logic Pro, Cubase/Nuendo, Studio One, and REAPER
• Introduce EDM-optimized analysis mode with enhanced BPM, key detection, frequency band, energy, and transient analysis, plus corresponding database schema extensions and CLI flags
• Add SQLite views creation and schema export command for analytical queries

Enhancements:

• Refactor FL Studio export to leverage generic metadata pipeline
• Unify CLI commands under 'export' and 'export-daw' with format, output path, and chunk size options
• Consolidate metadata building into a DAW-neutral module that standardizes feature normalization and tag inference

Documentation:

• Update README to reflect DAW-neutral pipeline, multi-format and DAW-specific exports, streaming support, and SQLite views
• Add EDM-optimized analysis documentation with feature details and usage examples

[sourcery-ai]

Reviewer's Guide

This PR restructures the export and analysis pipeline by centralizing metadata consolidation, introducing a generic DAW-neutral export module (with streaming), extending export formats (XML, Parquet, SQLite views), adding per-DAW adapters, integrating an EDM-optimized analysis flow (new schema, views, runner), overhauling the CLI for flexible commands, and updating documentation accordingly.

Sequence diagram for CLI command dispatch and export flow

sequenceDiagram
actor User
participant CLI
participant ExportGeneric
participant ExportExtended
participant ExportAbleton
participant ExportBitwig
participant ExportFL
participant ExportLogic
participant ExportCubase
participant ExportStudioOne
participant ExportReaper
User->>CLI: Run command (e.g. export, export-daw)
CLI->>ExportGeneric: run_export() (for DAW-neutral export)
CLI->>ExportExtended: run_export_xml()/run_export_parquet() (for extended formats)
CLI->>ExportAbleton: run_export_ableton() (for Ableton)
CLI->>ExportBitwig: run_export_bitwig() (for Bitwig)
CLI->>ExportFL: run_export_fl() (for FL Studio)
CLI->>ExportLogic: run_export_logic() (for Logic Pro)
CLI->>ExportCubase: run_export_cubase() (for Cubase)
CLI->>ExportStudioOne: run_export_studio_one() (for Studio One)
CLI->>ExportReaper: run_export_reaper() (for REAPER)
CLI-->>User: Print export result

Loading

Sequence diagram for EDM-optimized analysis flow

sequenceDiagram
actor User
participant CLI
participant AnalyzeEDMRunner
participant DBEDM
participant AnalyzeEDM
User->>CLI: Run analyze --edm
CLI->>DBEDM: add_edm_columns()
CLI->>AnalyzeEDMRunner: run_analyze_edm()
AnalyzeEDMRunner->>AnalyzeEDM: analyze_edm_features(file_path)
AnalyzeEDMRunner->>DBEDM: add_edm_columns(), create_edm_views()
AnalyzeEDMRunner->>DBEDM: Update features table with EDM columns
AnalyzeEDMRunner-->>CLI: Analysis complete
CLI-->>User: Print analysis result

Loading

Class diagram for new export and metadata modules

classDiagram
class Metadata {
  +build_sample_metadata(sample_id, path, relpath, duration, features)
  +export_all_metadata()
}
class ExportGeneric {
  +run_export(format, output_path)
  +run_export_streaming(format, output_path, chunk_size, show_progress)
}
class ExportExtended {
  +run_export_xml(output_path)
  +run_export_parquet(output_path)
  +run_create_sqlite_views()
  +export_sqlite_views_schema(output_path)
}
class ExportAbleton {
  +run_export_ableton(output_path)
}
class ExportBitwig {
  +run_export_bitwig(format, output_path)
}
class ExportFL {
  +run_export_fl(output_path, fl_user_data)
}
class ExportLogic {
  +run_export_logic(output_path)
}
class ExportCubase {
  +run_export_cubase(output_path)
}
class ExportStudioOne {
  +run_export_studio_one(output_path)
}
class ExportReaper {
  +run_export_reaper(format, output_path)
}
Metadata <|-- ExportGeneric
Metadata <|-- ExportExtended
Metadata <|-- ExportAbleton
Metadata <|-- ExportBitwig
Metadata <|-- ExportFL
Metadata <|-- ExportLogic
Metadata <|-- ExportCubase
Metadata <|-- ExportStudioOne
Metadata <|-- ExportReaper

Loading

Class diagram for EDM analysis and database extensions

classDiagram
class AnalyzeEDM {
  +analyze_edm_features(file_path)
  +detect_bpm_multipass(y, sr)
  +detect_key_enhanced(y, sr)
  +analyze_frequency_bands(y, sr)
  +calculate_energy_score(y, sr)
  +analyze_transients(y, sr)
}
class AnalyzeEDMRunner {
  +run_analyze_edm()
}
class DBEDM {
  +add_edm_columns()
  +create_edm_views()
}
AnalyzeEDMRunner --> AnalyzeEDM
AnalyzeEDMRunner --> DBEDM

Loading

File-Level Changes

Change	Details	Files
Centralize metadata consolidation and refactor FL export	Added metadata.py to build standardized metadata records and filename-regex parsing Removed legacy tag inference and DB queries from export_fl.py Refactored export_fl.py into convert_to_fl_format and export_to_fl_tags using export_all_metadata	src/metadata.py src/export_fl.py
Introduce generic DAW-neutral export with streaming support	Created export_generic.py for JSON/CSV/YAML exports and streaming export functions Updated run_pipeline.py to invoke the new generic export module Added CLI 'export' command with --format, --output, --streaming, and --chunk-size options	src/export_generic.py run_pipeline.py src/cli.py
Add extended export formats and SQLite views	Implemented export_extended.py for XML, Parquet, and SQLite views generation Introduced 'create-views' CLI command and --export-schema flag Wired XML and Parquet export commands into CLI under 'export' and 'export-daw'	src/export_extended.py src/cli.py
Implement DAW-specific adapter modules	Added export_ableton.py, export_bitwig.py, export_logic.py, export_cubase.py, export_studio_one.py, export_reaper.py Mapped each adapter in the new 'export-daw' CLI command Updated documentation to reflect 7 DAW formats supported	src/export_ableton.py src/export_bitwig.py src/export_logic.py src/export_cubase.py src/export_studio_one.py src/export_reaper.py src/cli.py
Integrate EDM-optimized analysis pipeline	Added analyze_edm.py with multi-pass BPM, Camelot key mapping, frequency, energy, and transient analysis Created db_edm.py to extend features schema and generate EDM-specific views Provided analyze_edm_runner.py and enhanced CLI flags (--edm, --setup-edm-db)	src/analyze_edm.py src/db_edm.py src/analyze_edm_runner.py src/cli.py
Overhaul CLI and update documentation	Replaced legacy export_fl command with unified 'export' and 'export-daw' subcommands Enhanced scan/analyze/autotype flows with new flags and EDM mode Rewrote README.md and added docs/EDM_ANALYSIS.md with updated features and usage	src/cli.py run_pipeline.py README.md docs/EDM_ANALYSIS.md

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[sourcery-ai]

Hey there - I've reviewed your changes - here's some feedback:

• The docstring for convert_to_fl_format says it returns a single string of file content, but the function actually returns a list of lines and a set of tags—please update the doc and signature to match the real return types.
• The export_to_fl_tags path‐building logic hardcodes backslashes and lowercases paths for Windows only; consider using pathlib.Path methods (e.g. relative_to, as_posix/os.sep) to build OS-agnostic paths.
• There’s a lot of repeated metadata–to–format conversion logic across the various DAW adapter modules; consider extracting a shared serialization utility or base class to reduce duplication and improve maintainability.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The docstring for convert_to_fl_format says it returns a single string of file content, but the function actually returns a list of lines and a set of tags—please update the doc and signature to match the real return types.
- The export_to_fl_tags path‐building logic hardcodes backslashes and lowercases paths for Windows only; consider using pathlib.Path methods (e.g. relative_to, as_posix/os.sep) to build OS-agnostic paths.
- There’s a lot of repeated metadata–to–format conversion logic across the various DAW adapter modules; consider extracting a shared serialization utility or base class to reduce duplication and improve maintainability.

## Individual Comments

### Comment 1
<location> `src/export_fl.py:15-24` </location>
<code_context>
+def convert_to_fl_format(metadata_list: list[dict[str, Any]], sample_roots: list[Path]) -> tuple[str, set[str]]:
</code_context>

<issue_to_address>
**issue:** The function signature claims to return a tuple[str, set[str]], but actually returns (list[str], set[str]).

Update the return type annotation to tuple[list[str], set[str]] for accurate type checking.
</issue_to_address>

### Comment 2
<location> `src/export_fl.py:43` </location>
<code_context>
+            all_tags.add(tag)
+
+        # Build FL-compatible path
+        if sample_roots and relpath:
+            base = Path(sample_roots[0]) if sample_roots else Path(path).drive + "\\"
+            lib_root_lower = str(Path(base)).lower().rstrip("\\/") + "\\"
</code_context>

<issue_to_address>
**suggestion:** The fallback for 'base' uses Path(path).drive + "\", which may not be robust for non-Windows paths.

Use Path(path).anchor or another cross-platform approach to reliably obtain the root directory.

```suggestion
            base = Path(sample_roots[0]) if sample_roots else Path(path).anchor
```
</issue_to_address>

### Comment 3
<location> `src/export_fl.py:77-81` </location>
<code_context>
-    for t in sorted(all_tags, key=lambda x: x.lower()):
-        if re.search(r'[,\s"]', t):
-            header += "," + '"' + t.replace('"', '') + '"'
+    for tag in sorted(all_tags, key=lambda x: x.lower()):
+        # Quote tags with special characters
+        if re.search(r'[,\s"]', tag):
+            header += "," + '"' + tag.replace('"', '') + '"'
</code_context>

<issue_to_address>
**suggestion (bug_risk):** The tag quoting logic removes double quotes but does not escape them, which could lead to malformed CSV if a tag contains a quote.

Escape double quotes within tags by replacing '"' with '""' to maintain valid CSV formatting.

```suggestion
        # Quote tags with special characters and escape double quotes
        if re.search(r'[,\s"]', tag):
            escaped_tag = tag.replace('"', '""')
            header += "," + f'"{escaped_tag}"'
        else:
            header += "," + tag
```
</issue_to_address>

### Comment 4
<location> `src/cli.py:123-124` </location>
<code_context>
+    if args.cmd == "export":
         try:
-            from .export_fl import run_export
+            if args.format in ["xml", "parquet"]:
+                from .export_extended import run_export_xml, run_export_parquet
+                output = Path(args.output) if args.output else None
+                if args.format == "xml":
</code_context>

<issue_to_address>
**suggestion:** The conditional for XML and Parquet formats is separated from the streaming logic, which could lead to confusion if streaming is later supported for these formats.

Please clarify in the CLI logic or documentation that streaming is not available for XML/Parquet formats.

Suggested implementation:

```python
    if args.cmd == "export":
        try:
            if args.format in ["xml", "parquet"]:
                if args.streaming:
                    print("Streaming is not supported for XML or Parquet export formats.")
                    return
                from .export_extended import run_export_xml, run_export_parquet
                output = Path(args.output) if args.output else None
                if args.format == "xml":
                    result_path = run_export_xml(output)
                else:  # parquet
                    result_path = run_export_parquet(output)
            elif args.streaming:
                from .export_generic import run_export_streaming
                output = Path(args.output) if args.output else None
                result_path = run_export_streaming(
                    format=args.format,
                    output_path=output,
                    chunk_size=args.chunk_size

```

If you have CLI argument definitions elsewhere (e.g., using argparse), update the help text for the `--streaming` flag to mention that streaming is not available for XML/Parquet formats. For example:

`parser.add_argument('--streaming', action='store_true', help='Enable streaming export (not available for XML or Parquet formats)')`
</issue_to_address>

### Comment 5
<location> `src/cli.py:130-131` </location>
<code_context>
+                    result_path = run_export_xml(output)
+                else:  # parquet
+                    result_path = run_export_parquet(output)
+            elif args.streaming:
+                from .export_generic import run_export_streaming
+                output = Path(args.output) if args.output else None
+                result_path = run_export_streaming(
</code_context>

<issue_to_address>
**issue (bug_risk):** The streaming export logic does not check if the selected format is supported for streaming.

Validate that the selected format supports streaming before starting the export to prevent runtime errors.
</issue_to_address>

### Comment 6
<location> `src/cli.py:197-198` </location>
<code_context>
+
+    if args.cmd == "create-views":
+        try:
+            from .export_extended import run_create_sqlite_views, export_sqlite_views_schema
+            views = run_create_sqlite_views()
+            print(f"Created {len(views)} SQLite views:")
+            for view_name in views.keys():
</code_context>

<issue_to_address>
**suggestion (bug_risk):** The CLI prints the number of views created but does not handle or report errors if some views fail to be created.

Please add error reporting for failed view creations to improve diagnostics.

Suggested implementation:

```python
            views, failed_views = run_create_sqlite_views()
            print(f"Created {len(views)} SQLite views:")
            for view_name in views.keys():
                print(f"  - {view_name}")

            if failed_views:
                print(f"\n[ERROR] Failed to create {len(failed_views)} views:")
                for view_name, error_msg in failed_views.items():
                    print(f"  - {view_name}: {error_msg}", file=sys.stderr)

            if args.export_schema:
                schema_path = export_sqlite_views_schema()
                print(f"\nSchema exported to: {schema_path}")

```

You will need to update the implementation of `run_create_sqlite_views()` in `export_extended.py` so that it returns a tuple `(views, failed_views)`, where `failed_views` is a dictionary mapping view names to error messages for any views that failed to be created.
</issue_to_address>

### Comment 7
<location> `src/export_generic.py:51-52` </location>
<code_context>
+        json.dump(metadata_list, f, indent=2, ensure_ascii=False)
+
+
+def export_to_csv(output_path: Path, metadata_list: list[dict]) -> None:
+    """
+    Export metadata to CSV format.
+
+    Args:
+        output_path: Output file path
+        metadata_list: List of metadata dictionaries
+    """
+    if not metadata_list:
+        return
+
+    # Flatten tags into comma-separated string
+    flattened = []
+    for item in metadata_list:
+        flat_item = item.copy()
+        flat_item["tags"] = ",".join(item.get("tags", []))
+        flattened.append(flat_item)
+
+    # Get all unique keys
+    fieldnames = list(flattened[0].keys())
+
+    with open(output_path, "w", encoding="utf-8", newline="") as f:
</code_context>

<issue_to_address>
**suggestion:** CSV fieldnames are determined from the first item, which may omit keys present in later items.

Collect all unique keys from every item to ensure the CSV header covers all fields, accommodating heterogeneous metadata.

```suggestion
    # Get all unique keys from every item
    fieldnames = set()
    for item in flattened:
        fieldnames.update(item.keys())
    fieldnames = list(fieldnames)
```
</issue_to_address>

### Comment 8
<location> `src/export_generic.py:118-127` </location>
<code_context>
+    return output_path
+
+
+def export_single_sample(sample_id: int, format: ExportFormat = "json") -> dict | str:
+    """
+    Export metadata for a single sample.
+
+    Args:
+        sample_id: Database ID of the sample
+        format: Output format
+
+    Returns:
+        Metadata as dict (json) or string (csv/yaml)
+    """
+    metadata_list = export_all_metadata()
+
+    # Find sample
+    sample_metadata = None
+    for item in metadata_list:
+        if item["sample_id"] == sample_id:
+            sample_metadata = item
+            break
+
+    if sample_metadata is None:
+        raise ValueError(f"Sample ID {sample_id} not found")
+
</code_context>

<issue_to_address>
**suggestion (performance):** The function loads all metadata to export a single sample, which is inefficient for large datasets.

Query only the necessary sample from the database to optimize performance and minimize memory usage.

Suggested implementation:

```python
def get_sample_metadata(sample_id: int) -> dict:
    """
    Query the database for metadata of a single sample.

    Args:
        sample_id: Database ID of the sample

    Returns:
        Metadata as dict

    Raises:
        ValueError: If sample not found
    """
    # Replace with actual database query logic
    # Example using SQLAlchemy:
    from src.models import Sample  # adjust import as needed
    sample = Sample.query.filter_by(id=sample_id).first()
    if sample is None:
        raise ValueError(f"Sample ID {sample_id} not found")
    return sample.to_dict()  # adjust to your ORM/model

def export_single_sample(sample_id: int, format: ExportFormat = "json") -> dict | str:
    """
    Export metadata for a single sample.

    Args:
        sample_id: Database ID of the sample
        format: Output format

    Returns:
        Metadata as dict (json) or string (csv/yaml)
    """
    sample_metadata = get_sample_metadata(sample_id)

```

- You may need to adjust the import and query logic in `get_sample_metadata` to match your database/ORM setup.
- If you already have a function to fetch a single sample's metadata, use that instead of implementing a new one.
- Update the rest of `export_single_sample` to handle formatting (json/csv/yaml) using `sample_metadata` directly.
</issue_to_address>

### Comment 9
<location> `src/metadata.py:173-174` </location>
<code_context>
+        tags.append(features["pred_type"])
+
+    # 2. Filename-derived tags
+    filename_tags = _parse_filename_tags(filename, regex_map)
+    for tag in filename_tags[:3]:  # Limit to top 3
+        if tag not in tags:
+            tags.append(tag)
</code_context>

<issue_to_address>
**suggestion:** Limiting filename-derived tags to the top 3 may omit relevant tags for some samples.

Please make the tag limit a configurable parameter, or add documentation explaining why three tags were chosen.

Suggested implementation:

```python
    # 2. Filename-derived tags
    # Limit for filename-derived tags is configurable via filename_tag_limit (default: 3).
    filename_tag_limit = 3  # Change this value or pass as a parameter to configure
    filename_tags = _parse_filename_tags(filename, regex_map)
    for tag in filename_tags[:filename_tag_limit]:
        if tag not in tags:
            tags.append(tag)

```

If you want the limit to be configurable from outside this function, you should:
1. Add `filename_tag_limit` as a parameter to the containing function.
2. Pass the desired value when calling this function elsewhere in your codebase.
3. Optionally, document the parameter in the function's docstring.
</issue_to_address>

### Comment 10
<location> `src/analyze_edm.py:48-53` </location>
<code_context>
def key_to_camelot(key: str | None) -> str | None:
    """
    Convert musical key to Camelot Wheel notation.

    Args:
        key: Musical key (e.g., "Am", "C", "F#m")

    Returns:
        Camelot notation (e.g., "8A", "8B")
    """
    if not key:
        return None

    # Normalize key
    key_normalized = key.strip()

    # Try direct lookup
    if key_normalized in CAMELOT_WHEEL:
        return CAMELOT_WHEEL[key_normalized]

    # Try with variations
    for k, v in CAMELOT_WHEEL.items():
        if key_normalized.upper() == k.upper():
            return v

    return None

</code_context>

<issue_to_address>
**suggestion (code-quality):** Use the built-in function `next` instead of a for-loop ([`use-next`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-next/))

```suggestion
    return next(
        (
            v
            for k, v in CAMELOT_WHEEL.items()
            if key_normalized.upper() == k.upper()
        ),
        None,
    )
```
</issue_to_address>

### Comment 11
<location> `src/analyze_edm.py:57` </location>
<code_context>
def get_compatible_keys(camelot: str) -> list[str]:
    """
    Get harmonically compatible keys for mixing.

    Args:
        camelot: Camelot notation (e.g., "8A")

    Returns:
        List of compatible Camelot keys
    """
    if not camelot or len(camelot) < 2:
        return []

    try:
        number = int(camelot[:-1])
        letter = camelot[-1].upper()
    except (ValueError, IndexError):
        return []

    compatible = []

    # Same key
    compatible.append(camelot)

    # +/- 1 on wheel (same letter)
    prev_num = number - 1 if number > 1 else 12
    next_num = number + 1 if number < 12 else 1
    compatible.append(f"{prev_num}{letter}")
    compatible.append(f"{next_num}{letter}")

    # Relative major/minor
    other_letter = "B" if letter == "A" else "A"
    compatible.append(f"{number}{other_letter}")

    return compatible

</code_context>

<issue_to_address>
**issue (code-quality):** We've found these issues:

- Merge append into list declaration ([`merge-list-append`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/merge-list-append/))
- Merge consecutive list appends into a single extend ([`merge-list-appends-into-extend`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/merge-list-appends-into-extend/))
- Move assignment closer to its usage within a block ([`move-assign-in-block`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/move-assign-in-block/))
- Merge extend into list declaration ([`merge-list-extend`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/merge-list-extend/))
</issue_to_address>

### Comment 12
<location> `src/analyze_edm.py:150-152` </location>
<code_context>
def detect_bpm_multipass(y: np.ndarray, sr: int) -> dict[str, Any]:
    """
    Multi-pass BPM detection optimized for EDM.

    Args:
        y: Audio time series
        sr: Sample rate

    Returns:
        Dictionary with BPM info and confidence
    """
    # Pass 1: HPSS separation
    try:
        y_harmonic, y_percussive = librosa.effects.hpss(y, margin=2.0)
    except Exception:
        y_percussive = y

    # Pass 2: Multiple tempo estimates
    tempo_estimates = []

    # Standard beat tracking on percussive
    tempo, beats = librosa.beat.beat_track(y=y_percussive, sr=sr, units='time')
    if tempo and tempo > 0:
        tempo_estimates.append(float(tempo))

    # Onset envelope method
    try:
        onset_env = librosa.onset.onset_strength(y=y_percussive, sr=sr)
        tempo_onset = librosa.feature.tempo(onset_envelope=onset_env, sr=sr)
        if len(tempo_onset) > 0 and tempo_onset[0] > 0:
            tempo_estimates.append(float(tempo_onset[0]))
    except Exception:
        pass

    # Tempogram method for more accuracy
    try:
        tempogram = librosa.feature.tempogram(y=y_percussive, sr=sr)
        tempo_tempogram = librosa.feature.tempo(onset_envelope=tempogram.mean(axis=0), sr=sr)
        if len(tempo_tempogram) > 0 and tempo_tempogram[0] > 0:
            tempo_estimates.append(float(tempo_tempogram[0]))
    except Exception:
        pass

    if not tempo_estimates:
        return {"bpm": None, "confidence": 0.0, "estimates": []}

    # Consensus voting
    tempo_median = float(np.median(tempo_estimates))
    tempo_std = float(np.std(tempo_estimates))

    # Confidence based on agreement
    confidence = 1.0 - min(tempo_std / tempo_median, 1.0) if tempo_median > 0 else 0.0

    # EDM-specific range optimization (most EDM is 110-180 BPM)
    candidates = [tempo_median / 2, tempo_median, tempo_median * 2]
    edm_candidates = [c for c in candidates if 110 <= c <= 180]

    if edm_candidates:
        final_bpm = min(edm_candidates, key=lambda c: abs(c - tempo_median))
    else:
        final_bpm = tempo_median

    return {
        "bpm": round(final_bpm, 1),
        "confidence": round(confidence, 3),
        "estimates": [round(t, 1) for t in tempo_estimates],
        "std_dev": round(tempo_std, 2)
    }

</code_context>

<issue_to_address>
**suggestion (code-quality):** Use named expression to simplify assignment and conditional ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))

```suggestion
    if edm_candidates := [c for c in candidates if 110 <= c <= 180]:
```
</issue_to_address>

### Comment 13
<location> `src/analyze_edm.py:324` </location>
<code_context>
def detect_key_enhanced(y: np.ndarray, sr: int) -> dict[str, Any]:
    """
    Enhanced key detection with longer analysis window.

    Args:
        y: Audio time series
        sr: Sample rate

    Returns:
        Dictionary with key info and confidence
    """
    # Use longer hop length for stability
    chroma = librosa.feature.chroma_cqt(y=y, sr=sr, hop_length=4096)

    # Average chroma over time
    chroma_mean = np.mean(chroma, axis=1)

    # Find dominant pitch class
    dominant_pitch = int(np.argmax(chroma_mean))

    # Determine major/minor (simplified heuristic)
    # Compare major vs minor profiles
    major_profile = np.array([1, 0, 1, 0, 1, 1, 0, 1, 0, 1, 0, 1])
    minor_profile = np.array([1, 0, 1, 1, 0, 1, 0, 1, 1, 0, 1, 0])

    # Rotate profiles to match dominant pitch
    major_rotated = np.roll(major_profile, dominant_pitch)
    minor_rotated = np.roll(minor_profile, dominant_pitch)

    # Correlation
    major_corr = float(np.corrcoef(chroma_mean, major_rotated)[0, 1])
    minor_corr = float(np.corrcoef(chroma_mean, minor_rotated)[0, 1])

    # Key names
    notes = ['C', 'C#', 'D', 'D#', 'E', 'F', 'F#', 'G', 'G#', 'A', 'A#', 'B']

    if major_corr > minor_corr:
        key = notes[dominant_pitch]
        confidence = major_corr
    else:
        key = notes[dominant_pitch] + 'm'
        confidence = minor_corr

    # Get Camelot notation
    camelot = key_to_camelot(key)
    compatible = get_compatible_keys(camelot) if camelot else []

    return {
        "key": key,
        "confidence": round(max(confidence, 0.0), 3),
        "camelot": camelot,
        "compatible_keys": compatible,
        "major_correlation": round(major_corr, 3),
        "minor_correlation": round(minor_corr, 3)
    }

</code_context>

<issue_to_address>
**suggestion (code-quality):** Use f-string instead of string concatenation ([`use-fstring-for-concatenation`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-fstring-for-concatenation/))

```suggestion
        key = f'{notes[dominant_pitch]}m'
```
</issue_to_address>

### Comment 14
<location> `src/analyze_edm.py:344` </location>
<code_context>
def analyze_edm_features(file_path: Path) -> dict[str, Any]:
    """
    Comprehensive EDM-optimized audio analysis.

    Args:
        file_path: Path to audio file

    Returns:
        Dictionary with all EDM-relevant features
    """
    # Load audio
    try:
        y, sr = librosa.load(file_path, sr=44100, mono=True)
    except Exception as e:
        return {"error": str(e)}

    features = {}

    # Enhanced BPM detection
    bpm_info = detect_bpm_multipass(y, sr)
    features["bpm"] = bpm_info["bpm"]
    features["bpm_confidence"] = bpm_info["confidence"]
    features["bpm_estimates"] = bpm_info["estimates"]
    features["bpm_std_dev"] = bpm_info.get("std_dev", 0.0)

    # Enhanced key detection
    key_info = detect_key_enhanced(y, sr)
    features["key"] = key_info["key"]
    features["key_confidence"] = key_info["confidence"]
    features["camelot"] = key_info["camelot"]
    features["compatible_keys"] = key_info["compatible_keys"]

    # Frequency band analysis
    freq_bands = analyze_frequency_bands(y, sr)
    features["frequency_bands"] = freq_bands

    # Energy analysis
    energy = calculate_energy_score(y, sr)
    features["energy"] = energy

    # Transient analysis
    transients = analyze_transients(y, sr)
    features["transients"] = transients

    # Duration
    features["duration"] = len(y) / sr

    return features

</code_context>

<issue_to_address>
**issue (code-quality):** We've found these issues:

- Move assignment closer to its usage within a block ([`move-assign-in-block`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/move-assign-in-block/))
- Merge dictionary assignment with declaration [×4] ([`merge-dict-assign`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/merge-dict-assign/))
</issue_to_address>

### Comment 15
<location> `src/analyze_edm_runner.py:41-47` </location>
<code_context>
def run_analyze_edm():
    """
    Run EDM-optimized analysis on all samples.
    """
    # Ensure EDM columns exist
    add_edm_columns()

    engine = init_db()

    # Get all samples
    with engine.begin() as conn:
        rows = conn.execute(text("""
            SELECT id, path FROM samples ORDER BY id
        """)).fetchall()

    print(f"[EDM] Analyzing {len(rows)} samples with enhanced precision...")

    for sample_id, path in tqdm(rows, desc="EDM Analysis"):
        try:
            # Run EDM analysis
            features = analyze_edm_features(Path(path))

            if "error" in features:
                continue

            # Update database with EDM features
            with engine.begin() as conn:
                # Check if feature row exists
                existing = conn.execute(
                    text("SELECT sample_id FROM features WHERE sample_id = :sid"),
                    dict(sid=sample_id)
                ).fetchone()

                if existing:
                    # Update existing row
                    conn.execute(text("""
                        UPDATE features SET
                            bpm = :bpm,
                            bpm_confidence = :bpm_conf,
                            key = :key,
                            key_conf = :key_conf,
                            camelot = :camelot,
                            sub_bass_energy = :sub_bass,
                            bass_energy = :bass,
                            mid_energy = :mid,
                            high_energy = :high,
                            energy_score = :energy_score,
                            transient_density = :trans_density,
                            dynamic_range = :dyn_range,
                            loudness = :loudness
                        WHERE sample_id = :sid
                    """), dict(
                        sid=sample_id,
                        bpm=features.get("bpm"),
                        bpm_conf=features.get("bpm_confidence"),
                        key=features.get("key"),
                        key_conf=features.get("key_confidence"),
                        camelot=features.get("camelot"),
                        sub_bass=features.get("frequency_bands", {}).get("sub_bass"),
                        bass=features.get("frequency_bands", {}).get("bass"),
                        mid=features.get("frequency_bands", {}).get("mid"),
                        high=features.get("frequency_bands", {}).get("high"),
                        energy_score=features.get("energy", {}).get("energy_score"),
                        trans_density=features.get("transients", {}).get("transient_density"),
                        dyn_range=features.get("energy", {}).get("dynamic_range"),
                        loudness=features.get("energy", {}).get("rms_mean")
                    ))
                else:
                    # Insert new row
                    conn.execute(text("""
                        INSERT INTO features (
                            sample_id, bpm, bpm_confidence, key, key_conf, camelot,
                            sub_bass_energy, bass_energy, mid_energy, high_energy,
                            energy_score, transient_density, dynamic_range, loudness
                        ) VALUES (
                            :sid, :bpm, :bpm_conf, :key, :key_conf, :camelot,
                            :sub_bass, :bass, :mid, :high,
                            :energy_score, :trans_density, :dyn_range, :loudness
                        )
                    """), dict(
                        sid=sample_id,
                        bpm=features.get("bpm"),
                        bpm_conf=features.get("bpm_confidence"),
                        key=features.get("key"),
                        key_conf=features.get("key_confidence"),
                        camelot=features.get("camelot"),
                        sub_bass=features.get("frequency_bands", {}).get("sub_bass"),
                        bass=features.get("frequency_bands", {}).get("bass"),
                        mid=features.get("frequency_bands", {}).get("mid"),
                        high=features.get("frequency_bands", {}).get("high"),
                        energy_score=features.get("energy", {}).get("energy_score"),
                        trans_density=features.get("transients", {}).get("transient_density"),
                        dyn_range=features.get("energy", {}).get("dynamic_range"),
                        loudness=features.get("energy", {}).get("rms_mean")
                    ))

        except Exception as e:
            print(f"\n[ERROR] Failed to analyze {path}: {e}")
            continue

    print(f"\n[EDM] Analysis complete. Enhanced features saved to database.")
    print(f"[EDM] Camelot keys, energy scores, and frequency bands available.")

</code_context>

<issue_to_address>
**suggestion (code-quality):** Use named expression to simplify assignment and conditional ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))

```suggestion
                if existing := conn.execute(
                    text(
                        "SELECT sample_id FROM features WHERE sample_id = :sid"
                    ),
                    dict(sid=sample_id),
                ).fetchone():
```
</issue_to_address>

### Comment 16
<location> `src/export_extended.py:103-108` </location>
<code_context>
def export_to_parquet(output_path: Path, metadata_list: list[dict[str, Any]]) -> Path:
    """
    Export metadata to Parquet format (requires pyarrow or fastparquet).

    Args:
        output_path: Output file path
        metadata_list: List of metadata dictionaries

    Returns:
        Path to created file
    """
    try:
        import pandas as pd
    except ImportError:
        raise ImportError(
            "pandas is required for Parquet export. Install with: pip install pandas pyarrow"
        )

    # Flatten tags for DataFrame
    flattened = []
    for item in metadata_list:
        flat_item = item.copy()
        flat_item["tags"] = "|".join(item.get("tags", []))  # Use | as separator
        flattened.append(flat_item)

    # Create DataFrame
    df = pd.DataFrame(flattened)

    # Write to Parquet
    df.to_parquet(output_path, index=False, engine="pyarrow")

    return output_path

</code_context>

<issue_to_address>
**issue (code-quality):** Explicitly raise from a previous error ([`raise-from-previous-error`](https://docs.sourcery.ai/Reference/Default-Rules/suggestions/raise-from-previous-error/))
</issue_to_address>

### Comment 17
<location> `src/export_fl.py:51` </location>
<code_context>
def convert_to_fl_format(metadata_list: list[dict[str, Any]], sample_roots: list[Path]) -> tuple[str, set[str]]:
    """
    Convert generic metadata to FL Studio Browser Tags format.

    FL Studio uses a CSV-like format with a header defining all tags,
    followed by rows with file paths and their tags.

    Args:
        metadata_list: List of generic metadata dictionaries
        sample_roots: List of sample root directories

    Returns:
        Tuple of (tags_file_content, all_tags_set)
    """
    all_tags = set()
    lines = []

    for item in metadata_list:
        path = item.get("path", "")
        relpath = item.get("relpath", "")
        tags = item.get("tags", [])

        # Add all tags to global set
        for tag in tags:
            all_tags.add(tag)

        # Build FL-compatible path
        if sample_roots and relpath:
            base = Path(sample_roots[0]) if sample_roots else Path(path).drive + "\\"
            lib_root_lower = str(Path(base)).lower().rstrip("\\/") + "\\"
            final_path = lib_root_lower + relpath.replace("/", "\\")
        else:
            final_path = path

        # Build line: "path",tag1,tag2,tag3
        tag_str = ",".join(tags) if tags else ""
        lines.append(f'"{final_path}"' + ("," + tag_str if tag_str else ""))

    return lines, all_tags

</code_context>

<issue_to_address>
**suggestion (code-quality):** Use f-string instead of string concatenation ([`use-fstring-for-concatenation`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-fstring-for-concatenation/))

```suggestion
        lines.append(f'"{final_path}"' + (f",{tag_str}" if tag_str else ""))
```
</issue_to_address>

### Comment 18
<location> `src/export_fl.py:81` </location>
<code_context>
def export_to_fl_tags(
    output_path: Path,
    metadata_list: list[dict[str, Any]],
    sample_roots: list[Path]
) -> Path:
    """
    Export to FL Studio Browser Tags format.

    Args:
        output_path: Output file path
        metadata_list: List of metadata dictionaries
        sample_roots: List of sample root directories

    Returns:
        Path to created file
    """
    lines, all_tags = convert_to_fl_format(metadata_list, sample_roots)

    # Build header: @TagCase=*,Tag1,Tag2,Tag3,...
    header = "@TagCase=*"
    for tag in sorted(all_tags, key=lambda x: x.lower()):
        # Quote tags with special characters
        if re.search(r'[,\s"]', tag):
            header += "," + '"' + tag.replace('"', '') + '"'
        else:
            header += "," + tag

    # Write file
    with open(output_path, "w", encoding="utf-8") as f:
        f.write(header + "\n")
        for line in lines:
            f.write(line + "\n")

    return output_path

</code_context>

<issue_to_address>
**suggestion (code-quality):** Use f-string instead of string concatenation ([`use-fstring-for-concatenation`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-fstring-for-concatenation/))

```suggestion
            header += f",{tag}"
```
</issue_to_address>

### Comment 19
<location> `src/export_generic.py:68-73` </location>
<code_context>
def export_to_yaml(output_path: Path, metadata_list: list[dict]) -> None:
    """
    Export metadata to YAML format (requires PyYAML).

    Args:
        output_path: Output file path
        metadata_list: List of metadata dictionaries
    """
    try:
        import yaml
    except ImportError:
        raise ImportError(
            "PyYAML is required for YAML export. Install with: pip install pyyaml"
        )

    with open(output_path, "w", encoding="utf-8") as f:
        yaml.dump(metadata_list, f, default_flow_style=False, allow_unicode=True)

</code_context>

<issue_to_address>
**issue (code-quality):** Explicitly raise from a previous error ([`raise-from-previous-error`](https://docs.sourcery.ai/Reference/Default-Rules/suggestions/raise-from-previous-error/))
</issue_to_address>

### Comment 20
<location> `src/export_generic.py:131-137` </location>
<code_context>
def export_single_sample(sample_id: int, format: ExportFormat = "json") -> dict | str:
    """
    Export metadata for a single sample.

    Args:
        sample_id: Database ID of the sample
        format: Output format

    Returns:
        Metadata as dict (json) or string (csv/yaml)
    """
    metadata_list = export_all_metadata()

    # Find sample
    sample_metadata = None
    for item in metadata_list:
        if item["sample_id"] == sample_id:
            sample_metadata = item
            break

    if sample_metadata is None:
        raise ValueError(f"Sample ID {sample_id} not found")

    if format == "json":
        return sample_metadata
    elif format == "csv":
        # Return as CSV row
        flat = sample_metadata.copy()
        flat["tags"] = ",".join(sample_metadata.get("tags", []))
        return ",".join(str(v) for v in flat.values())
    elif format == "yaml":
        import yaml
        return yaml.dump([sample_metadata], default_flow_style=False, allow_unicode=True)
    else:
        raise ValueError(f"Unsupported format: {format}")

</code_context>

<issue_to_address>
**suggestion (code-quality):** Use the built-in function `next` instead of a for-loop ([`use-next`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-next/))

```suggestion
    sample_metadata = next(
        (item for item in metadata_list if item["sample_id"] == sample_id),
        None,
    )
```
</issue_to_address>

### Comment 21
<location> `src/metadata.py:74-78` </location>
<code_context>
def _normalize_key(key: str | None, confidence: float | None) -> str | None:
    """Normalize key notation to standard format."""
    if not key or (confidence is not None and confidence < CONF_KEY_MIN):
        return None

    # Standardize notation
    k = key.replace("min", "m").replace("maj", "").upper()
    if len(k) == 1:
        k = k + "maj"
    if k.endswith("M"):
        k = k[:-1] + "maj"
    if k.endswith("m"):
        k = k[:-1] + "min"
    return k

</code_context>

<issue_to_address>
**issue (code-quality):** Use f-string instead of string concatenation [×3] ([`use-fstring-for-concatenation`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-fstring-for-concatenation/))
</issue_to_address>

### Comment 22
<location> `src/metadata.py:84-86` </location>
<code_context>
def _normalize_bpm(bpm: float | None) -> int | None:
    """Normalize BPM to integer."""
    if not bpm or bpm <= 0:
        return None
    return int(round(bpm))

</code_context>

<issue_to_address>
**suggestion (code-quality):** We've found these issues:

- Lift code into else after jump in control flow ([`reintroduce-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/reintroduce-else/))
- Replace if statement with if expression ([`assign-if-exp`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/assign-if-exp/))

```suggestion
    return None if not bpm or bpm <= 0 else int(round(bpm))
```
</issue_to_address>

### Comment 23
<location> `src/metadata.py:95-97` </location>
<code_context>
def _classify_brightness(brightness: float | None) -> str | None:
    """Classify brightness into categories."""
    if brightness is None:
        return None
    if brightness < 1500:
        return "Dark"
    if brightness > 3500:
        return "Bright"
    return None

</code_context>

<issue_to_address>
**suggestion (code-quality):** We've found these issues:

- Lift code into else after jump in control flow ([`reintroduce-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/reintroduce-else/))
- Replace if statement with if expression ([`assign-if-exp`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/assign-if-exp/))

```suggestion
    return "Bright" if brightness > 3500 else None
```
</issue_to_address>

### Comment 24
<location> `src/metadata.py:106-108` </location>
<code_context>
def _classify_loudness(loudness: float | None) -> str | None:
    """Classify loudness into categories."""
    if loudness is None:
        return None
    if loudness > -18:
        return "Punchy"
    if loudness < -28:
        return "Clean"
    return None

</code_context>

<issue_to_address>
**suggestion (code-quality):** We've found these issues:

- Lift code into else after jump in control flow ([`reintroduce-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/reintroduce-else/))
- Replace if statement with if expression ([`assign-if-exp`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/assign-if-exp/))

```suggestion
    return "Clean" if loudness < -28 else None
```
</issue_to_address>

### Comment 25
<location> `src/metadata.py:115-117` </location>
<code_context>
def _classify_duration(duration: float | None, clazz: str | None) -> str | None:
    """Classify duration class."""
    if clazz == "oneshot":
        return "OneShot"
    if clazz == "loop":
        return "Loop"
    return None

</code_context>

<issue_to_address>
**suggestion (code-quality):** We've found these issues:

- Lift code into else after jump in control flow ([`reintroduce-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/reintroduce-else/))
- Replace if statement with if expression ([`assign-if-exp`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/assign-if-exp/))

```suggestion
    return "Loop" if clazz == "loop" else None
```
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[sourcery-ai]

issue: The function signature claims to return a tuple[str, set[str]], but actually returns (list[str], set[str]).

Update the return type annotation to tuple[list[str], set[str]] for accurate type checking.

[sourcery-ai]

suggestion: The fallback for 'base' uses Path(path).drive + "", which may not be robust for non-Windows paths.

Use Path(path).anchor or another cross-platform approach to reliably obtain the root directory.

Suggested change

	base = Path(sample_roots[0]) if sample_roots else Path(path).drive + "\\"
	base = Path(sample_roots[0]) if sample_roots else Path(path).anchor

[sourcery-ai]

suggestion (bug_risk): The tag quoting logic removes double quotes but does not escape them, which could lead to malformed CSV if a tag contains a quote.

Escape double quotes within tags by replacing '"' with '""' to maintain valid CSV formatting.

Suggested change

	# Quote tags with special characters
	if re.search(r'[,\s"]', tag):
	header += "," + '"' + tag.replace('"', '') + '"'
	else:
	header += "," + t
	header += "," + tag
	# Quote tags with special characters and escape double quotes
	if re.search(r'[,\s"]', tag):
	escaped_tag = tag.replace('"', '""')
	header += "," + f'"{escaped_tag}"'
	else:
	header += "," + tag

[sourcery-ai]

suggestion: The conditional for XML and Parquet formats is separated from the streaming logic, which could lead to confusion if streaming is later supported for these formats.

Please clarify in the CLI logic or documentation that streaming is not available for XML/Parquet formats.

Suggested implementation:

    if args.cmd == "export":
        try:
            if args.format in ["xml", "parquet"]:
                if args.streaming:
                    print("Streaming is not supported for XML or Parquet export formats.")
                    return
                from .export_extended import run_export_xml, run_export_parquet
                output = Path(args.output) if args.output else None
                if args.format == "xml":
                    result_path = run_export_xml(output)
                else:  # parquet
                    result_path = run_export_parquet(output)
            elif args.streaming:
                from .export_generic import run_export_streaming
                output = Path(args.output) if args.output else None
                result_path = run_export_streaming(
                    format=args.format,
                    output_path=output,
                    chunk_size=args.chunk_size

If you have CLI argument definitions elsewhere (e.g., using argparse), update the help text for the --streaming flag to mention that streaming is not available for XML/Parquet formats. For example:

parser.add_argument('--streaming', action='store_true', help='Enable streaming export (not available for XML or Parquet formats)')

[sourcery-ai]

issue (bug_risk): The streaming export logic does not check if the selected format is supported for streaming.

Validate that the selected format supports streaming before starting the export to prevent runtime errors.

[sourcery-ai]

suggestion (code-quality): Use the built-in function next instead of a for-loop (use-next)

Suggested change

	# Find sample
	sample_metadata = None
	for item in metadata_list:
	if item["sample_id"] == sample_id:
	sample_metadata = item
	break
	sample_metadata = next(
	(item for item in metadata_list if item["sample_id"] == sample_id),
	None,
	)

[sourcery-ai]

suggestion (code-quality): We've found these issues:

• Lift code into else after jump in control flow (reintroduce-else)
• Replace if statement with if expression (assign-if-exp)

Suggested change

	if not bpm or bpm <= 0:
	return None
	return int(round(bpm))
	return None if not bpm or bpm <= 0 else int(round(bpm))

[sourcery-ai]

suggestion (code-quality): We've found these issues:

• Lift code into else after jump in control flow (reintroduce-else)
• Replace if statement with if expression (assign-if-exp)

Suggested change

	if brightness > 3500:
	return "Bright"
	return None
	return "Bright" if brightness > 3500 else None

[sourcery-ai]

suggestion (code-quality): We've found these issues:

• Lift code into else after jump in control flow (reintroduce-else)
• Replace if statement with if expression (assign-if-exp)

Suggested change

	if loudness < -28:
	return "Clean"
	return None
	return "Clean" if loudness < -28 else None

[sourcery-ai]

suggestion (code-quality): We've found these issues:

• Lift code into else after jump in control flow (reintroduce-else)
• Replace if statement with if expression (assign-if-exp)

Suggested change

	if clazz == "loop":
	return "Loop"
	return None
	return "Loop" if clazz == "loop" else None