# Data Monitor

Comprehensive data monitoring system for analyzing large-scale training/validation datasets with parallel chunk processing, smart caching, and modular analyzers.

## Overview

The data monitor analyzes JSONL datasets with 57M+ records efficiently using parallel chunk processing. It provides detailed statistics on IDs, tags, text patterns, languages, stems, file paths, and more.

## Quick Start

### Setup

```bash
cd sunoData/src/sunodata/data_monitor

# Install dependencies (if needed)
pip install -r requirements.txt
```

### Basic Usage

```bash
# Test on 100k records (fast)
python test_pipeline.py --dataset v9 --sample-size 100000

# Full analysis of a dataset
python run_pipeline.py --dataset v9 --full

# Analyze specific splits
python run_pipeline.py --dataset v9 --split train
python run_pipeline.py --dataset v9 --split val

# Run specific analyzers only
python run_pipeline.py --dataset v9 --analyzers tags id text
```

## Output Locations

All outputs are stored outside the repository:

- **Cache**: `~/data/suno_data_monitor/cache/` - Line counts, ID sets, analysis results
- **Temporary**: `~/data/suno_data_monitor/tmp/` - Intermediate processing files
- **Outputs**: `~/data/suno_data_monitor/outputs/` - Analysis results and reports
- **Samples**: `~/data/suno_data_monitor/outputs/{run}/samples/` - Individual example records

## Analyzers

### 1. ID Analyzer (`id`)
Analyzes ID uniqueness, duplicates, and overlaps between datasets.

**Outputs:**
- Unique ID count
- Duplicate detection
- Train/val overlap statistics
- ID length distributions

### 2. Tags Analyzer (`tags`)
Analyzes tag distributions and discovers numeric value_tags.

**Outputs:**
- Tag count distributions
- Discovered value_tags (with `:` followed by numeric values)
- Known value_tags: `quality`, `bass`, `mid`, `high`, `loudness`, `spectral_centroid`, etc.
- Tag length statistics

### 3. Text Analyzer (`text`)
Analyzes text patterns, lengths, and bracket usage.

**Outputs:**
- Text length distributions
- Bracket pattern detection: `[]`, `{}`, `()`, `【】`, `《》`, `「」`, `『』`
- Bracket position analysis (start/middle/end)
- Content length within brackets

### 4. Language Analyzer (`language`)
Analyzes language distribution and detects mismatches.

**Outputs:**
- Language frequency distribution
- Mismatch detection (declared vs detected language)
- Language-text combinations
- Supported: 100+ languages including CJK, Cyrillic, Arabic, etc.

### 5. Weight Analyzer (`weight`)
Analyzes weight value distributions.

**Outputs:**
- Weight distribution statistics
- Histogram of weight values
- Min/max weights

### 6. Stems Analyzer (`stems`)
Analyzes stem tracks and captions.

**Outputs:**
- Stem count distributions
- Stem name frequencies (Vocals, Drums, Bass, etc.)
- Stems_captions statistics

### 7. Paths Analyzer (`paths`)
Analyzes file path patterns.

**Outputs:**
- Local filepath patterns (checks `/app2/suno/data/`)
- S3 filepath patterns (bucket/category analysis)
- File extensions
- Path anomalies

### 8. Lists Analyzer (`lists`)
Analyzes list fields and their combinations.

**Outputs:**
- `artist_ids`, `cover_ids`, `playlist_ids`, `stems` count distributions
- **Compound combinations**: e.g., `artist_playlist`, `artist_cover_playlist_stems`
- Percentage breakdown of all field combinations

### 9. Sample Collector (`samples`)
Collects interesting example records for inspection.

**Captures:**
- Max total tag length
- Longest single tag
- Most tags
- Shortest/longest text
- Examples for each language (85+)
- Examples for each stem type (12)
- Examples with different bracket types
- Extreme weights/durations
- Multi-artist/cover records
- Records with special fields (vocal_pitch_range, text_aligned)

**Output:** Individual JSON files in `samples/` directory

### 10. SFT Analyzer (`sft`)
Validates SFT (supervised fine-tuning) ID subsets.

**Outputs:**
- SFT subset sizes
- Missing IDs validation
- Coverage statistics

## Configuration

Edit `config.yaml` to configure datasets and processing parameters:

```yaml
datasets:
  v9:
    train: /app2/suno/data/auk_v0/metas_v9_tr.jsonl
    val: /app2/suno/data/auk_v0/metas_v8_val.jsonl
    sft_ids: /app2/suno/data/auk_v0/ids_keep_sets_v13.json

processing:
  chunk_size: 2_000_000      # Records per chunk
  parallel_workers: 8         # Number of parallel workers
  test_size: 100_000         # Size for test runs
```

## Advanced Usage

### Clear Cache

```bash
python run_pipeline.py --dataset v9 --clear-cache --full
```

### Disable Cache

```bash
python run_pipeline.py --dataset v9 --no-cache --split train
```

### Test Parallel Consistency

```bash
# Run both sequential and parallel, then compare
python test_pipeline.py --dataset v9 --sample-size 10000

# Skip comparison (faster)
python test_pipeline.py --dataset v9 --sample-size 50000 --skip-comparison

# Skip parallel test (sequential only)
python test_pipeline.py --dataset v9 --sample-size 100000 --skip-parallel
```

### Custom Workers and Chunk Size

```bash
python test_pipeline.py --dataset v9 --sample-size 100000 --workers 16 --chunk-size 5000
```

## Performance

- **Sequential**: ~3.9s for 100k records
- **Parallel (8 workers)**: ~0.6s for 100k records (6.5x speedup)
- **Full dataset (57M records)**: Estimated ~30-45 minutes with caching

## Output Format

All analysis results are saved as JSON files with the following structure:

```json
{
  "summary": {
    "total_records": 100000,
    "field_specific_stats": "..."
  },
  "detailed_analysis": {
    "distributions": "...",
    "top_values": "..."
  }
}
```

Sample files use `indent=2` for readability.

## Architecture

### Parallel Processing
- Splits files into 20+ chunks (configurable)
- Each chunk processes ALL analyzers sequentially
- Chunks processed in parallel across multiple workers
- Results aggregated deterministically

### Caching
- Line counts cached with file modification time
- ID sets cached as pickle files
- Analysis results cached per file/analyzer
- Auto-invalidation on file changes

### Modularity
- Each analyzer is independent
- Easy to add new analyzers
- Consistent interface: `process_chunk()` → `aggregate()`

## Troubleshooting

### Out of Memory
Reduce `chunk_size` or `parallel_workers` in `config.yaml`:
```yaml
processing:
  chunk_size: 1_000_000
  parallel_workers: 4
```

### Cache Issues
Clear cache if results seem stale:
```bash
python run_pipeline.py --dataset v9 --clear-cache
```

### Missing Dependencies
```bash
pip install pyyaml tqdm numpy
```

## Examples

### Compare Dataset Versions

```bash
# Analyze v8 and v9 to compare
python run_pipeline.py --dataset v8 --full
python run_pipeline.py --dataset v9 --full

# Check outputs in ~/data/suno_data_monitor/outputs/
```

### Investigate Specific Fields

```bash
# Deep dive into tags only
python run_pipeline.py --dataset v9 --analyzers tags

# Check language distribution
python run_pipeline.py --dataset v9 --analyzers lang

# Validate IDs and check overlaps
python run_pipeline.py --dataset v9 --analyzers id
```

### Collect Samples

```bash
# Run with sample collector enabled (included by default)
python test_pipeline.py --dataset v9 --sample-size 50000 --skip-parallel

# Check samples directory
ls ~/data/suno_data_monitor/outputs/test_v9_*/samples/
```

## Development

### Adding a New Analyzer

1. Create `analyzers/new_analyzer.py`:
```python
from .base_analyzer import BaseAnalyzer

class NewAnalyzer(BaseAnalyzer):
    def __init__(self, output_dir):
        super().__init__('new', output_dir)

    def process_chunk(self, records):
        # Process chunk, return intermediate results
        return {'intermediate': 'data'}

    def aggregate(self, chunk_results):
        # Aggregate all chunks
        return {'final': 'results'}
```

2. Add to `chunk_processor.py`:
```python
from ..analyzers.new_analyzer import NewAnalyzer

analyzer_classes = {
    # ...
    'new': NewAnalyzer
}
```

3. Add to pipeline scripts (test_pipeline.py, run_pipeline.py)

### Testing

```bash
# Test on small sample first
python test_pipeline.py --dataset v9 --sample-size 1000

# Verify consistency
python test_pipeline.py --dataset v9 --sample-size 10000
```

## Notes

- Cache and output directories are automatically created outside the repository
- All file processing uses UTF-8 encoding (`ensure_ascii=False`)
- Results are deterministic and consistent between runs
- Sample collector captures up to 85 languages, all stem types, and extremes