Add Cursor rules

.cursor/rules/coding-standards.mdc

@@ -0,0 +1,84 @@
+---
+alwaysApply: true
+---
+
+# NeMo Curator Coding Standards
+
+## Linting and Formatting
+
+The project uses **Ruff** for linting and formatting with line length of 119 characters.
+
+## Key Style Rules
+
+### Allowed Patterns
+
+- ✅ Print statements (T20 ignored)
+- ✅ Boolean arguments in functions (FBT ignored)
+- ✅ `df` as variable name for DataFrames (PD901 ignored)
+- ✅ TODOs without author/link (TD002, TD003 ignored)
+- ✅ Long exception messages (TRY003 ignored)
+- ✅ Accessing private attributes (SLF001 ignored)
+- ✅ Branching after return (RET505-508 ignored)
+
+### Required Patterns
+
+- ❌ No docstrings required (D ignored)
+- ❌ No pathlib enforcement (PTH ignored)
+- ❌ No logging enforcement (G ignored)
+- ✅ Type annotations for functions (except `*args`, `**kwargs`, special methods)
+
+## File-Specific Exceptions
+
+### `examples/` directory
+- No `__init__.py` required (INP001)
+
+### `tests/` directory
+- Allow assertions (S101)
+- Allow magic values (PLR2004)
+- No return type annotations required (ANN201)
+
+### `tutorials/` directory
+- No `__init__.py` required (INP001)
+- Ignore Unicode complaint (PLE2515)
+
+## Copyright Header
+
+All non-empty Python files must include the NVIDIA copyright header:
+
+```python
+# Copyright (c) 2025, NVIDIA CORPORATION.  All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+```
+
+## Python Version Support
+
+Supports Python 3.10, 3.11, and 3.12 (requires >=3.10, <3.13)
+
+## Loguru
+
+NeMo Curator uses loguru for logging:
+
+```python
+from loguru import logger
+```
+
+Common uses include `logger.info`, `logger.warning`, and `logger.error`.
+
+## PyTest Standards
+
+All changes to NeMo Curator's source code must be accompanied with relevant tests. Tests which require a GPU and cannot be run on a CPU-only machine must be marked with `@pytest.mark.gpu`.
+
+NeMo Curator enforces 80% PyTest coverage within the `nemo_curator/` directory.
+
+For straightforward navigation purposes, the `tests/` directory structure matches the `nemo_curator/` directory structure.

.cursor/rules/composite-stage-patterns.mdc

@@ -0,0 +1,70 @@
+---
+alwaysApply: true
+---
+
+# `CompositeStage` Patterns
+
+## When to Use `CompositeStage`
+
+Use `CompositeStage` for high-level, user-facing stages that decompose into multiple low-level execution stages:
+
+- Provide simplified API while maintaining fine-grained execution control
+- Bundle multiple related stages into a single logical operation
+- Especially useful when stages require different resources (e.g., a CPU-based stage followed by a GPU-based stage)
+- Decomposed during pipeline planning
+
+## Creating a `CompositeStage`
+
+```python
+from dataclasses import dataclass
+
+from nemo_curator.stages.base import CompositeStage, ProcessingStage
+
+
+@dataclass
+class MyCompositeStage(CompositeStage[InputTaskType, OutputTaskType]):
+    param1: str
+    param2: int
+
+    def __post_init__(self) -> None:
+        super().__init__()
+
+        self.stages = [
+            StageA(param1=self.param1),
+            StageB(param2=self.param2),
+            StageC(),
+        ]
+
+    def inputs(self) -> tuple[list[str], list[str]]:
+        # StageA's inputs
+        return self.stages[0].inputs()
+
+    def outputs(self) -> tuple[list[str], list[str]]:
+        # StageC's outputs
+        return self.stages[2].outputs()
+
+    def decompose(self) -> list[ProcessingStage]:
+        return self.stages
+```
+
+## Configuration with `with_()`
+
+`CompositeStage` use a different `with_()` signature that takes a dictionary:
+
+```python
+from nemo_curator.stages.resources import Resources
+
+
+composite_stage = MyCompositeStage(param1, param2)
+
+# Add a with operation
+stage_config = {"StageA": {"resources": Resources(cpus=5.0)}}
+updated_composite_stage = composite_stage.with_(stage_config)
+```
+
+## Important Rules
+
+- Decomposed stages cannot be `CompositeStage`s themselves
+- `inputs()` returns first stage's inputs
+- `outputs()` returns last stage's outputs
+- All stages in `decompose()` must have unique names for `with_()` to work

.cursor/rules/executors.mdc

@@ -0,0 +1,98 @@
+---
+alwaysApply: true
+---
+
+# Executors
+
+Executors are the runtime engines that execute NeMo Curator pipelines. They handle distributed task orchestration, resource allocation, and worker management.
+
+## Role of Executors
+
+Executors translate the high-level pipeline definition (a sequence of `ProcessingStage` instances) into actual distributed execution:
+
+- **Orchestrate stage execution**: Run stages in sequence, passing tasks between them
+- **Distribute work**: Parallelize task processing across workers/nodes
+- **Manage resources**: Allocate CPUs, GPUs, memory according to stage requirements
+- **Handle batching**: Group tasks into batches for efficient processing
+- **Track performance**: Collect timing and throughput metrics
+- **Manage lifecycle**: Setup and teardown workers/resources
+
+## Available Executors
+
+### `XennaExecutor` (Default)
+
+Production executor using Cosmos-Xenna for distributed execution:
+
+```python
+from nemo_curator.backends.xenna import XennaExecutor
+
+executor = XennaExecutor(config={
+    "logging_interval": 60,
+    "ignore_failures": False,
+    "execution_mode": "streaming",  # or "batch"
+    "cpu_allocation_percentage": 0.95,
+    "autoscale_interval_s": 180,
+})
+```
+
+### Experimental Executors
+
+Located in `nemo_curator.backends.experimental`:
+
+- **RayDataExecutor**: Ray Data backend (supports `ignore_head_node`)
+- **RayActorPoolExecutor**: Ray Actor Pool backend (supports `ignore_head_node`)
+
+## `BaseExecutor` Interface
+
+All executors inherit from `BaseExecutor` and implement:
+
+```python
+class BaseExecutor(ABC):
+    def __init__(self, config: dict[str, Any] | None = None, ignore_head_node: bool = False):
+        """Initialize executor with configuration.
+
+        Args:
+            config: Executor-specific configuration dictionary
+            ignore_head_node: Whether to exclude head node from execution (not supported by XennaExecutor)
+        """
+
+    @abstractmethod
+    def execute(self, stages: list[ProcessingStage], initial_tasks: list[Task] | None = None) -> list[Task]:
+        """Execute the pipeline stages.
+
+        Args:
+            stages: List of processing stages to execute
+            initial_tasks: Initial tasks to start pipeline (defaults to EmptyTask)
+
+        Returns:
+            List of output tasks from final stage
+        """
+```
+
+## Stage Adapters
+
+Executors use `BaseStageAdapter` to wrap stages for execution:
+
+- Implements batching logic via `process_batch()`
+- Calls stage lifecycle methods (`setup_on_node()`, `setup()`, `teardown()`)
+- Tracks performance metrics with `StageTimer`
+- Attaches performance statistics to output tasks
+
+## Usage in Pipelines
+
+Executors are passed to `pipeline.run()`:
+
+```python
+from nemo_curator.pipeline import Pipeline
+from nemo_curator.backends.xenna import XennaExecutor
+
+
+pipeline = Pipeline(name="my_pipeline", stages=[...])
+
+# Explicit executor
+executor = XennaExecutor(config={"execution_mode": "streaming"})
+results = pipeline.run(executor=executor)
+
+# Default executor (XennaExecutor)
+results = pipeline.run()
+```

.cursor/rules/modality-structure.mdc

@@ -0,0 +1,70 @@
+---
+alwaysApply: true
+---
+
+# Modality Structure
+
+NeMo Curator supports four data modalities, each with its own stage directory and patterns.
+
+## Directory Structure
+
+```
+nemo_curator/stages/
+├── text/ # Text/document processing
+├── image/ # Image processing
+├── audio/ # Audio/speech processing
+└── video/ # Video processing
+```
+
+Other subdirectories of `nemo_curator/stages/` are `deduplication` and `synthetic`, which contain stages that can be used by two or more modalities.
+
+## Text Processing (`stages/text/`)
+
+Common subdirectories:
+- `classifiers/`: Quality, domain, content safety classification
+- `deduplication/`: Duplicate text removal
+- `download/`: Data ingestion (Common Crawl, Wikipedia, ArXiv)
+- `embedders/`: Text embedding generation
+- `filters/`: Heuristic and model-based filtering
+- `io/`: Readers (Parquet, JSONL) and writers
+- `modifiers/`: Text transformations (cleaning, normalization)
+
+Task type: `DocumentBatch`
+
+## Image Processing (`stages/image/`)
+
+Common operations:
+- CLIP embedding generation
+- Aesthetic quality scoring
+- NSFW detection
+- Deduplication
+
+Task type: `ImageBatch`
+
+## Audio Processing (`stages/audio/`)
+
+Common operations:
+- ASR transcription (NeMo Framework)
+- Word Error Rate (WER) calculation
+- Duration analysis
+- Quality filtering
+
+Task type: `AudioBatch`
+
+## Video Processing (`stages/video/`)
+
+Common operations:
+- Scene detection (TransNetV2)
+- Clip extraction
+- GPU H.264 encoding/decoding
+- Motion and aesthetic filtering
+- Embeddings (InternVideo2, Cosmos-Embed1)
+
+Task type: `VideoTask`
+
+## Shared Components
+
+All modalities share:
+- `stages/base.py`: Base classes (`ProcessingStage`, `CompositeStage`)
+- `stages/resources.py`: Resource configuration
+- `stages/function_decorators.py`: Decorators for creating `ProcessingStage` instances from simple functions

.cursor/rules/pipeline-structure.mdc

@@ -0,0 +1,57 @@
+---
+alwaysApply: true
+---
+
+# Pipeline Structure
+
+## Creating Pipelines
+
+Pipelines compose `ProcessingStage` instances into an executable workflow:
+
+```python
+from nemo_curator.pipeline import Pipeline
+
+pipeline = Pipeline(
+    name="my_pipeline",
+    description="My data processing pipeline",
+    stages=[stage1, stage2, stage3],
+)
+```
+
+## Adding Stages
+
+Stages can be added during initialization or via `add_stage()`:
+
+```python
+# Method 1: During initialization
+pipeline = Pipeline(name="my_pipeline", stages=[stage1, stage2])
+
+# Method 2: Add stages individually
+pipeline = Pipeline(name="my_pipeline")
+pipeline.add_stage(stage1)
+pipeline.add_stage(stage2)
+```
+
+## Running Pipelines
+
+```python
+pipeline.run()
+```
+
+The `run()` method accepts 2 optional parameters:
+
+- `executor`: Executor to use. If None, defaults to `XennaExecutor`
+- `initial_tasks` Initial `Task`s to start the pipeline with. Defaults to None.
+
+Before executing, the `run()` method calls `self.build()` to build an execution plan from the pipeline (e.g., decomposes composite stages).
+
+## Composite Stage Decomposition
+
+When a pipeline is built, any `CompositeStage` instances are automatically decomposed into their constituent execution stages. The decomposition info is stored in `pipeline.decomposition_info`.
+
+## Pipeline Methods
+
+- `add_stage(stage)`: Add a stage to the pipeline (returns self for chaining)
+- `build()`: Decompose composite stages into execution stages
+- `describe()`: Get detailed description of pipeline stages and requirements
+- `run(executor, initial_tasks)`: Execute the pipeline