Apply ruff formatting to QwenImage warning implementation

- Fix whitespace and string quote consistency

- Add trailing commas where appropriate

- Clean up formatting per diffusers code standards

Author: robin-ede

src/diffusers/models/transformers/transformer_qwenimage.py

@@ -164,22 +164,28 @@ def __init__(self, theta: int, axes_dim: List[int], scale_rope=False):
         self._current_max_len = 1024
         pos_index = torch.arange(self._current_max_len)
         neg_index = torch.arange(self._current_max_len).flip(0) * -1 - 1
-        self.register_buffer('pos_freqs', torch.cat(
-            [
-                self.rope_params(pos_index, self.axes_dim[0], self.theta),
-                self.rope_params(pos_index, self.axes_dim[1], self.theta),
-                self.rope_params(pos_index, self.axes_dim[2], self.theta),
-            ],
-            dim=1,
-        ))
-        self.register_buffer('neg_freqs', torch.cat(
-            [
-                self.rope_params(neg_index, self.axes_dim[0], self.theta),
-                self.rope_params(neg_index, self.axes_dim[1], self.theta),
-                self.rope_params(neg_index, self.axes_dim[2], self.theta),
-            ],
-            dim=1,
-        ))
+        self.register_buffer(
+            "pos_freqs",
+            torch.cat(
+                [
+                    self.rope_params(pos_index, self.axes_dim[0], self.theta),
+                    self.rope_params(pos_index, self.axes_dim[1], self.theta),
+                    self.rope_params(pos_index, self.axes_dim[2], self.theta),
+                ],
+                dim=1,
+            ),
+        )
+        self.register_buffer(
+            "neg_freqs",
+            torch.cat(
+                [
+                    self.rope_params(neg_index, self.axes_dim[0], self.theta),
+                    self.rope_params(neg_index, self.axes_dim[1], self.theta),
+                    self.rope_params(neg_index, self.axes_dim[2], self.theta),
+                ],
+                dim=1,
+            ),
+        )
         self.rope_cache = {}
 
         # 是否使用 scale rope
@@ -199,22 +205,22 @@ def _expand_pos_freqs_if_needed(self, required_len):
         """Expand pos_freqs and neg_freqs if required length exceeds current size"""
         if required_len <= self._current_max_len:
             return
-        
+
         # Calculate new size (use next power of 2 or round to nearest 512 for efficiency)
         new_max_len = max(required_len, int((required_len + 511) // 512) * 512)
-        
+
         # Log warning about potential quality degradation for long prompts
         if required_len > 512:
             logger.warning(
                 f"QwenImage model was trained on prompts up to 512 tokens. "
                 f"Current prompt requires {required_len} tokens, which may lead to unpredictable behavior. "
                 f"Consider using shorter prompts for better results."
             )
-        
+
         # Generate expanded indices
         pos_index = torch.arange(new_max_len, device=self.pos_freqs.device)
         neg_index = torch.arange(new_max_len, device=self.neg_freqs.device).flip(0) * -1 - 1
-        
+
         # Generate expanded frequency embeddings
         new_pos_freqs = torch.cat(
             [
@@ -224,7 +230,7 @@ def _expand_pos_freqs_if_needed(self, required_len):
             ],
             dim=1,
         ).to(device=self.pos_freqs.device, dtype=self.pos_freqs.dtype)
-        
+
         new_neg_freqs = torch.cat(
             [
                 self.rope_params(neg_index, self.axes_dim[0], self.theta),
@@ -233,12 +239,12 @@ def _expand_pos_freqs_if_needed(self, required_len):
             ],
             dim=1,
         ).to(device=self.neg_freqs.device, dtype=self.neg_freqs.dtype)
-        
+
         # Update buffers
-        self.register_buffer('pos_freqs', new_pos_freqs)
-        self.register_buffer('neg_freqs', new_neg_freqs)
+        self.register_buffer("pos_freqs", new_pos_freqs)
+        self.register_buffer("neg_freqs", new_neg_freqs)
         self._current_max_len = new_max_len
-        
+
         # Clear cache since dimensions changed
         self.rope_cache = {}
 
@@ -281,11 +287,11 @@ def forward(self, video_fhw, txt_seq_lens, device):
             max_vid_index = max(height, width)
 
         max_len = max(txt_seq_lens)
-        
+
         # Expand pos_freqs if needed to accommodate max_vid_index + max_len
         required_len = max_vid_index + max_len
         self._expand_pos_freqs_if_needed(required_len)
-        
+
         txt_freqs = self.pos_freqs[max_vid_index : max_vid_index + max_len, ...]
 
         return vid_freqs, txt_freqs

tests/pipelines/qwenimage/test_qwenimage.py

@@ -241,43 +241,43 @@ def test_long_prompt_no_error(self):
         components = self.get_dummy_components()
         pipe = self.pipeline_class(**components)
         pipe.to(device)
-        
+
         # Create a very long prompt that exceeds 1024 tokens when combined with image positioning
         # Repeat a long phrase to simulate a real long prompt scenario
         long_phrase = "A beautiful, detailed, high-resolution, photorealistic image showing "
         long_prompt = (long_phrase * 50)[:1200]  # Ensure we exceed 1024 characters
-        
+
         inputs = {
             "prompt": long_prompt,
             "generator": torch.Generator(device=device).manual_seed(0),
             "num_inference_steps": 2,
             "guidance_scale": 3.0,
             "true_cfg_scale": 1.0,
             "height": 32,  # Small size for fast test
-            "width": 32,   # Small size for fast test
+            "width": 32,  # Small size for fast test
             "max_sequence_length": 1200,  # Allow long sequence
             "output_type": "pt",
         }
-        
+
         # This should not raise a RuntimeError about tensor dimension mismatch
         _ = pipe(**inputs)
 
     def test_long_prompt_warning(self):
         """Test that long prompts trigger appropriate warning about training limitation"""
         from diffusers.utils import logging
-        
+
         components = self.get_dummy_components()
         pipe = self.pipeline_class(**components)
         pipe.to(torch_device)
-        
+
         # Create prompt that will exceed 512 tokens to trigger warning
         long_phrase = "A detailed photorealistic description of a complex scene with many elements "
         long_prompt = (long_phrase * 20)[:800]  # Create a prompt that will exceed 512 tokens
-        
-        # Capture transformer logging  
+
+        # Capture transformer logging
         logger = logging.get_logger("diffusers.models.transformers.transformer_qwenimage")
         logger.setLevel(logging.WARNING)
-        
+
         with CaptureLogger(logger) as cap_logger:
             _ = pipe(
                 prompt=long_prompt,
@@ -286,11 +286,11 @@ def test_long_prompt_warning(self):
                 guidance_scale=3.0,
                 true_cfg_scale=1.0,
                 height=32,  # Small size for fast test
-                width=32,   # Small size for fast test
+                width=32,  # Small size for fast test
                 max_sequence_length=900,  # Allow long sequence
-                output_type="pt"
+                output_type="pt",
             )
-        
+
         # Verify warning was logged about the 512-token training limitation
         self.assertTrue("512 tokens" in cap_logger.out)
         self.assertTrue("unpredictable behavior" in cap_logger.out)