[diffusion] Clean up duplicate helper definitions (#30159)

python/sglang/multimodal_gen/configs/pipeline_configs/base.py

@@ -222,9 +222,6 @@ class PipelineConfig:
     text_encoder_precisions: tuple[str, ...] = field(default_factory=lambda: ("fp32",))
     text_encoder_extra_args: list[dict] = field(default_factory=lambda: [{}])
 
-    def get_model_deployment_config(self) -> ModelDeploymentConfig:
-        return ModelDeploymentConfig()
-
     def postprocess_image(self, image):
         return image.last_hidden_state
 
@@ -289,9 +286,6 @@ class PipelineConfig:
             (target_width, target_height), PIL.Image.Resampling.LANCZOS
         ), (target_width, target_height)
 
-    def preprocess_realtime_condition_image(self, batch, _vae_image_processor) -> bool:
-        return False
-
     def prepare_calculated_size(self, image):
         return self.calculate_condition_image_size(image, image.width, image.height)
 

python/sglang/multimodal_gen/runtime/pipelines_core/stages/model_specific_stages/joy_echo/memory.py

@@ -320,11 +320,6 @@ def build_memory_self_attention_block_mask(
 # --- Memory VAE encode ---
 
 
-import torch
-from PIL import Image
-from torchvision.transforms import functional as TVF
-
-
 def frames_to_video_tensor(
     frames: list[Image.Image], target_h: int, target_w: int
 ) -> torch.Tensor:
@@ -512,11 +507,6 @@ def build_memory_audio_rope_coords(
 # --- Paired audio-video memory bank ---
 
 
-import torch
-import torchaudio
-from PIL import Image
-
-
 @dataclass
 class MemoryEntry:
     frame: Image.Image | list[Image.Image]

python/sglang/multimodal_gen/test/scripts/gen_diffusion_ci_outputs.py

@@ -13,7 +13,6 @@ Usage:
 
 import argparse
 import os
-import subprocess
 import sys
 from pathlib import Path
 
@@ -26,53 +25,15 @@ from sglang.multimodal_gen.test.run_suite import (
     get_suite_files_rel,
     parse_partition_plan,
     partition_items_by_lpt,
+)
+from sglang.multimodal_gen.test.runner.pytest_runner import (
+    collect_test_items,
     run_pytest,
 )
-from sglang.multimodal_gen.test.runner.pytest_runner import collect_test_items
 
 logger = init_logger(__name__)
 
 
-def collect_test_items(files: list[str], filter_expr: str | None = None) -> list[str]:
-    """Collect test node IDs from the given files using pytest --collect-only."""
-    cmd = [sys.executable, "-m", "pytest", "--collect-only", "-q"]
-    if filter_expr:
-        cmd.extend(["-k", filter_expr])
-    cmd.extend(files)
-
-    filter_note = f" with filter: {filter_expr}" if filter_expr else ""
-    print(f"Collecting tests from {len(files)} file(s){filter_note}")
-    result = subprocess.run(cmd, capture_output=True, text=True)
-
-    if result.returncode not in (0, 5):
-        error_msg = (
-            f"pytest --collect-only failed with exit code {result.returncode}\n"
-            f"Command: {' '.join(cmd)}\n"
-        )
-        if result.stderr:
-            error_msg += f"stderr:\n{result.stderr}\n"
-        if result.stdout:
-            error_msg += f"stdout:\n{result.stdout}\n"
-        logger.error(error_msg)
-        raise RuntimeError(error_msg)
-
-    if result.returncode == 5:
-        print(
-            "No tests were collected (exit code 5). This may be expected with filters."
-        )
-
-    test_items = []
-    for line in result.stdout.strip().split("\n"):
-        line = line.strip()
-        if line and "::" in line and not line.startswith(("=", "-", " ")):
-            test_id = line.split()[0] if " " in line else line
-            if "::" in test_id:
-                test_items.append(test_id)
-
-    print(f"Collected {len(test_items)} test items")
-    return test_items
-
-
 def main():
     """Main entry point."""
     parser = argparse.ArgumentParser(description="Generate diffusion CI outputs")