fix(MediaStreamInput): ensure all frames are processed in order

- Process first frame immediately after reading for dimensions
- Use explicit index-based iteration (range(1, num_frames)) for remaining frames
- Eliminates iterator position ambiguity that could skip frame 0 or read it twice

media_stream.py

@@ -8,13 +8,15 @@ import imageio.v2 as imageio
 import mimetypes
 import boto3
 import json
+
+import tempfile
+import os
 from .logger import logger
 from .config.config import load_nilor_nodes_config
 
 # Load shared configuration once
 _CFG = load_nilor_nodes_config()
 
-
 # --- Node Categories ---
 category = "Nilor Nodes 👺"
 subcategories = {
@@ -54,10 +56,10 @@ class MediaStreamInput:
     CATEGORY = category + subcategories["streaming"]
 
     def download(
-        self,
-        presigned_download_url: str,
-        format: str,
-        input_name: str = "default_input",
+            self,
+            presigned_download_url: str,
+            format: str,
+            input_name: str = "default_input",
     ):
         logger.info(
             f"ℹ️\u2009 Nilor-Nodes: MediaStreamInput: Downloading from {presigned_download_url} for input '{input_name}' with format '{format}'"
@@ -94,19 +96,35 @@ class MediaStreamInput:
                 return self._process_image_batch(asset_responses)
 
             # --- Single-file download ---
-            response = requests.get(presigned_download_url, timeout=180)
-            response.raise_for_status()
-            media_bytes = response.content
-
             if format == "video":
-                return self._process_video(media_bytes)
-            elif format == "image":
-                return self._process_image(media_bytes)
+                # Stream video to temp file to avoid loading entire video into RAM
+                temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
+                try:
+                    logger.info(
+                        f"ℹ️\u2009 Nilor-Nodes (MediaStreamInput): Streaming video to temp file: {temp_file.name}")
+                    with requests.get(presigned_download_url, timeout=180, stream=True) as response:
+                        response.raise_for_status()
+                        for chunk in response.iter_content(chunk_size=8192):
+                            temp_file.write(chunk)
+                    temp_file.close()
+                    return self._process_video(temp_file.name)
+                finally:
+                    # Clean up temp file
+                    if os.path.exists(temp_file.name):
+                        os.unlink(temp_file.name)
             else:
-                # Should not happen if UI choices are respected
-                raise ValueError(
-                    f"🛑\u2009 Nilor-Nodes (MediaStreamInput): Unsupported format '{format}' for single media download."
-                )
+                # For images, load into memory (they're small)
+                response = requests.get(presigned_download_url, timeout=180)
+                response.raise_for_status()
+                media_bytes = response.content
+
+                if format == "image":
+                    return self._process_image(media_bytes)
+                else:
+                    # Should not happen if UI choices are respected
+                    raise ValueError(
+                        f"🛑\u2009 Nilor-Nodes (MediaStreamInput): Unsupported format '{format}' for single media download."
+                    )
 
         except requests.RequestException as e:
             logger.error(
@@ -156,27 +174,45 @@ class MediaStreamInput:
         logger.info("✅ Nilor-Nodes (MediaStreamInput): Image processing successful.")
         return (image_tensor,)
 
-    def _process_video(self, video_bytes):
-        logger.info("ℹ️\u2009 Nilor-Nodes (MediaStreamInput): Processing as video...")
-        frames = []
-        with imageio.get_reader(io.BytesIO(video_bytes), format="mp4") as reader:
-            for frame in reader:
-                # Convert frame to RGB PIL Image and then to tensor
-                pil_image = Image.fromarray(frame).convert("RGB")
-                numpy_image = np.array(pil_image).astype(np.float32) / 255.0
-                tensor_frame = torch.from_numpy(numpy_image)
-                frames.append(tensor_frame)
+    def _process_video(self, video_path):
+        logger.info(f"ℹ️\u2009 Nilor-Nodes (MediaStreamInput): Processing video from {video_path}...")
 
-        if not frames:
-            raise ValueError(
-                "🛑\u2009 Nilor-Nodes (MediaStreamInput): No frames could be read from the video."
+        # Open video to get metadata first
+        with imageio.get_reader(video_path, format="mp4") as reader:
+            # Get video metadata
+            metadata = reader.get_meta_data()
+            num_frames = reader.count_frames()
+
+            if num_frames == 0:
+                raise ValueError(
+                    "🛑\u2009 Nilor-Nodes (MediaStreamInput): No frames could be read from the video."
+                )
+
+            # Read first frame to get dimensions
+            first_frame = reader.get_data(0)
+            height, width = first_frame.shape[:2]
+
+            logger.info(
+                f"ℹ️\u2009 Nilor-Nodes (MediaStreamInput): Video has {num_frames} frames at {width}x{height}"
             )
 
-        # Stack frames into a single tensor (batch of images)
-        video_tensor = torch.stack(frames)
+            # Pre-allocate tensor for all frames (N, H, W, 3)
+            video_tensor = torch.empty((num_frames, height, width, 3), dtype=torch.float32)
 
-        logging.info(
-            f"✅ Nilor-Nodes (MediaStreamInput): Video processing successful. Image Shape: {video_tensor.shape}"
+            # Process first frame (already read for dimensions)
+            pil_image = Image.fromarray(first_frame).convert("RGB")
+            numpy_image = np.array(pil_image).astype(np.float32) / 255.0
+            video_tensor[0] = torch.from_numpy(numpy_image)
+
+            # Read remaining frames by explicit index to avoid iterator position ambiguity
+            for i in range(1, num_frames):
+                frame = reader.get_data(i)
+                pil_image = Image.fromarray(frame).convert("RGB")
+                numpy_image = np.array(pil_image).astype(np.float32) / 255.0
+                video_tensor[i] = torch.from_numpy(numpy_image)
+
+        logger.info(
+            f"✅ Nilor-Nodes (MediaStreamInput): Video processing successful. Tensor shape: {video_tensor.shape}"
         )
         return (video_tensor,)
 
@@ -240,21 +276,21 @@ class MediaStreamOutput:
     CATEGORY = category + subcategories["streaming"]
 
     def upload_and_notify(
-        self,
-        images,
-        format,
-        content_id,
-        venue,
-        canvas,
-        scene,
-        presigned_upload_url,
-        job_completions_queue_url,
-        output_object_keys,
-        framerate,
-        output_name: str = "default_output",
-        prompt=None,
-        extra_pnginfo=None,
-        job_type: str | None = None,
+            self,
+            images,
+            format,
+            content_id,
+            venue,
+            canvas,
+            scene,
+            presigned_upload_url,
+            job_completions_queue_url,
+            output_object_keys,
+            framerate,
+            output_name: str = "default_output",
+            prompt=None,
+            extra_pnginfo=None,
+            job_type: str | None = None,
     ):
         if not content_id:
             raise ValueError(

user_input.py

@@ -3,9 +3,6 @@ subcategories = {
     "io": "/IO",
 }
 
-import random
-from datetime import datetime
-
 from .controllers import CONTROLLER_HOOK
 
 
@@ -53,83 +50,6 @@ class NilorUserInput_Int:
         return (value, None)
 
 
-class NilorUserInput_Seed:
-    MAX_COMFYUI_SEED = 1125899906842624
-    SEED_RANDOM_STATE = None
-
-    @classmethod
-    def _ensure_seed_random_state(cls):
-        if cls.SEED_RANDOM_STATE is not None:
-            return
-
-        initial_random_state = random.getstate()
-        random.seed(datetime.now().timestamp())
-        cls.SEED_RANDOM_STATE = random.getstate()
-        random.setstate(initial_random_state)
-
-    @classmethod
-    def generate_random_seed(cls):
-        cls._ensure_seed_random_state()
-
-        prev_random_state = random.getstate()
-        random.setstate(cls.SEED_RANDOM_STATE)
-        seed = random.randint(0, cls.MAX_COMFYUI_SEED)
-        cls.SEED_RANDOM_STATE = random.getstate()
-        random.setstate(prev_random_state)
-        return seed
-
-    @classmethod
-    def resolve_seed(cls, value):
-        if value in (None, 0, -1):
-            return cls.generate_random_seed()
-        try:
-            return int(value) % (cls.MAX_COMFYUI_SEED + 1)
-        except (TypeError, ValueError):
-            return cls.generate_random_seed()
-
-    @classmethod
-    def INPUT_TYPES(cls):
-        return {
-            "required": {
-                "input_name": (
-                    "STRING",
-                    {"default": "my_seed_input", "multiline": False},
-                ),
-                "value": (
-                    "INT",
-                    {
-                        "default": -1,
-                        "min": -1,
-                        "max": cls.MAX_COMFYUI_SEED,
-                    },
-                ),
-            },
-            "hidden": {
-                "prompt": "PROMPT",
-                "extra_pnginfo": "EXTRA_PNGINFO",
-                "unique_id": "UNIQUE_ID",
-            },
-        }
-
-    RETURN_TYPES = ("INT", CONTROLLER_HOOK)
-    RETURN_NAMES = ("seed", "_controller_hook")
-    FUNCTION = "get_value"
-    CATEGORY = category + subcategories["io"]
-
-    @classmethod
-    def IS_CHANGED(
-        cls, input_name, value, prompt=None, extra_pnginfo=None, unique_id=None
-    ):
-        # Force node re-execution while using randomize sentinel values.
-        return cls.resolve_seed(value)
-
-    def get_value(
-        self, input_name, value, prompt=None, extra_pnginfo=None, unique_id=None
-    ):
-        value = self.resolve_seed(value)
-        return (value, None)
-
-
 class NilorUserInput_Float:
     @classmethod
     def INPUT_TYPES(cls):
@@ -177,7 +97,6 @@ class NilorUserInput_Boolean:
 NODE_CLASS_MAPPINGS = {
     "NilorUserInput_String": NilorUserInput_String,
     "NilorUserInput_Int": NilorUserInput_Int,
-    "NilorUserInput_Seed": NilorUserInput_Seed,
     "NilorUserInput_Float": NilorUserInput_Float,
     "NilorUserInput_Boolean": NilorUserInput_Boolean,
 }
@@ -185,7 +104,6 @@ NODE_CLASS_MAPPINGS = {
 NODE_DISPLAY_NAME_MAPPINGS = {
     "NilorUserInput_String": "👺 User Input (String)",
     "NilorUserInput_Int": "👺 User Input (Int)",
-    "NilorUserInput_Seed": "👺 User Input (Seed)",
     "NilorUserInput_Float": "👺 User Input (Float)",
     "NilorUserInput_Boolean": "👺 User Input (Boolean)",
 }

worker_consumer.py

@@ -29,10 +29,6 @@ from .config.config import load_nilor_nodes_config, NilorNodesConfig
 _CFG: NilorNodesConfig = load_nilor_nodes_config()
 
 
-class JobSubmissionError(Exception):
-    """Raised when a job cannot be submitted to local ComfyUI."""
-
-
 class WorkerConsumer:
     def __init__(self, cfg: NilorNodesConfig):
         self.session = get_session()
@@ -401,19 +397,7 @@ class WorkerConsumer:
                 return
 
             # Submit to ComfyUI
-            try:
-                await self._submit_job_to_comfyui(content_id, job_payload)
-            except JobSubmissionError as e:
-                logger.error(
-                    f"🛑\u2009 Nilor-Nodes (worker_consumer): Submission failed for content_id {content_id}: {e}. Message will be retried/DLQ'd."
-                )
-                await self._emit_failed_status_for_submission_error(
-                    content_id=content_id,
-                    job_payload=job_payload,
-                    error_message=str(e),
-                )
-                # Re-raise so consume_loop does not delete the message.
-                raise
+            await self._submit_job_to_comfyui(content_id, job_payload)
 
             # Cache context for subsequent status updates
             try:
@@ -434,28 +418,6 @@ class WorkerConsumer:
             # Re-raise to prevent deletion from queue if we want SQS to handle retry
             raise
 
-    async def _emit_failed_status_for_submission_error(
-        self, content_id, job_payload, error_message: str
-    ):
-        """Best-effort failed status emission for submit-time errors."""
-        policy = job_payload.get("status_policy") or {}
-        fail_status = policy.get("fail_status", "failed")
-
-        await self._send_status_update(
-            content_id,
-            fail_status,
-            job_payload.get("venue"),
-            job_payload.get("canvas"),
-            job_payload.get("scene"),
-            job_payload.get("job_type"),
-        )
-        logger.info(
-            "ℹ️\u2009 Nilor-Nodes (worker_consumer): Emitted failed status '%s' for content_id %s after submit error: %s",
-            fail_status,
-            content_id,
-            error_message,
-        )
-
     async def _submit_job_to_comfyui(self, content_id, workflow_data):
         """Submits a single job to the ComfyUI API."""
         try:
@@ -522,18 +484,15 @@ class WorkerConsumer:
             logger.error(
                 f"🛑\u2009 Nilor-Nodes (worker_consumer): Failed to submit job to ComfyUI: {e}. Message will be retried."
             )
-            raise JobSubmissionError(str(e)) from e
         except (json.JSONDecodeError, KeyError) as e:
             logger.error(
-                f"🛑\u2009 Nilor-Nodes (worker_consumer): Failed to parse ComfyUI response: {e}. Message will be retried."
+                f"🛑\u2009 Nilor-Nodes (worker_consumer): Failed to parse ComfyUI response: {e}. Discarding malformed response."
             )
-            raise JobSubmissionError(f"Malformed ComfyUI response: {e}") from e
         except Exception as e:
             logger.error(
                 f"🛑\u2009 Nilor-Nodes (worker_consumer): An unexpected error occurred while submitting job to ComfyUI: {e}",
                 exc_info=True,
             )
-            raise JobSubmissionError(str(e)) from e
 
     async def _send_status_update(
         self, content_id, status, venue=None, canvas=None, scene=None, job_type=None