diff --git a/api/job_routes.py b/api/job_routes.py index d962adf..b35655b 100644 --- a/api/job_routes.py +++ b/api/job_routes.py @@ -295,17 +295,24 @@ async def job_complete_endpoint(request): errors.append("worker_id: expected non-empty string") if not isinstance(batch_idx, int) or batch_idx < 0: errors.append("batch_idx: expected non-negative integer") - if not isinstance(image_payload, str) or not image_payload.strip(): + image_provided = image_payload is not None + audio_provided = audio_payload is not None + if not image_provided and not audio_provided: + errors.append("expected at least one of image or audio") + if image_provided and (not isinstance(image_payload, str) or not image_payload.strip()): errors.append("image: expected non-empty base64 PNG string") - if audio_payload is not None and not isinstance(audio_payload, dict): + if audio_provided and not isinstance(audio_payload, dict): errors.append("audio: expected object when provided") if not isinstance(is_last, bool): errors.append("is_last: expected boolean") if errors: return await handle_api_error(request, errors, 400) - tensor = _decode_canonical_png_tensor(image_payload) - decoded_audio = _decode_audio_payload(audio_payload) if audio_payload is not None else None + try: + tensor = _decode_canonical_png_tensor(image_payload) if image_provided else None + decoded_audio = _decode_audio_payload(audio_payload) if audio_provided else None + except ValueError as exc: + return await handle_api_error(request, exc, 400) multi_job_id = job_id.strip() worker_id = worker_id.strip() diff --git a/api/orchestration/prompt_transform.py b/api/orchestration/prompt_transform.py index ebaa29f..ef25de8 100644 --- a/api/orchestration/prompt_transform.py +++ b/api/orchestration/prompt_transform.py @@ -351,16 +351,33 @@ def prune_prompt_for_worker(prompt_obj): downstream = _find_downstream_nodes(prompt_obj, [dist_id]) has_removed_downstream = any(node_id != dist_id for node_id in downstream) if has_removed_downstream: - preview_id = next_id() - pruned_prompt[preview_id] = { - "inputs": { - "images": [dist_id, 0], - }, - "class_type": "PreviewImage", - "_meta": { - "title": "Preview Image (auto-added)", - }, - } + original_node = prompt_obj.get(str(dist_id), {}) + class_type = original_node.get("class_type") + inputs = original_node.get("inputs", {}) + image_connected = class_type != "DistributedCollector" or ( + isinstance(inputs.get("images"), list) + and len(inputs["images"]) == 2 + ) + audio_connected = ( + class_type == "DistributedCollector" + and isinstance(inputs.get("audio"), list) + and len(inputs["audio"]) == 2 + ) + + if image_connected: + preview_id = next_id() + pruned_prompt[preview_id] = { + "inputs": {"images": [dist_id, 0]}, + "class_type": "PreviewImage", + "_meta": {"title": "Preview Image (auto-added)"}, + } + elif audio_connected: + preview_id = next_id() + pruned_prompt[preview_id] = { + "inputs": {"audio": [dist_id, 1]}, + "class_type": "PreviewAudio", + "_meta": {"title": "Preview Audio (auto-added)"}, + } return pruned_prompt @@ -400,6 +417,10 @@ def prepare_delegate_master_prompt(prompt_obj, collector_ids): collector_entry = pruned_prompt.get(collector_id) if not collector_entry: continue + original_inputs = (prompt_obj.get(collector_id) or {}).get("inputs", {}) + original_images = original_inputs.get("images") + if not (isinstance(original_images, list) and len(original_images) == 2): + continue placeholder_id = next_id() pruned_prompt[placeholder_id] = { "class_type": "DistributedEmptyImage", diff --git a/nodes/collector.py b/nodes/collector.py index 4528985..d577b0f 100644 --- a/nodes/collector.py +++ b/nodes/collector.py @@ -29,7 +29,6 @@ class DistributedCollectorNode: def INPUT_TYPES(s): return { "required": { - "images": ("IMAGE",), "load_balance": ( "BOOLEAN", { @@ -38,7 +37,10 @@ class DistributedCollectorNode: }, ), }, - "optional": { "audio": ("AUDIO",) }, + "optional": { + "images": ("IMAGE",), + "audio": ("AUDIO",), + }, "hidden": { "multi_job_id": ("STRING", {"default": ""}), "is_worker": ("BOOLEAN", {"default": False}), @@ -102,8 +104,9 @@ class DistributedCollectorNode: return None return {"waveform": torch.cat(waveforms, dim=-1), "sample_rate": sample_rate} - def run(self, images, load_balance=False, audio=None, multi_job_id="", is_worker=False, master_url="", enabled_worker_ids="[]", worker_batch_size=1, worker_id="", pass_through=False, delegate_only=False): - images = self._normalize_images_input(images) + def run(self, images=None, load_balance=False, audio=None, multi_job_id="", is_worker=False, master_url="", enabled_worker_ids="[]", worker_batch_size=1, worker_id="", pass_through=False, delegate_only=False): + if images is not None: + images = self._normalize_images_input(images) audio = self._normalize_audio_input(audio) load_balance = self._unwrap_list_input(load_balance) multi_job_id = self._unwrap_list_input(multi_job_id) @@ -115,6 +118,14 @@ class DistributedCollectorNode: pass_through = self._unwrap_list_input(pass_through) delegate_only = self._unwrap_list_input(delegate_only) + remote_only_master = ( + bool(multi_job_id) + and not is_worker + and (delegate_only or is_master_delegate_only()) + ) + if images is None and audio is None and not remote_only_master: + raise ValueError("DistributedCollector requires at least one image or audio input") + # Create empty audio if not provided empty_audio = {"waveform": torch.zeros(1, 2, 1), "sample_rate": 44100} @@ -141,41 +152,56 @@ class DistributedCollectorNode: return result async def send_batch_to_master(self, image_batch, audio, multi_job_id, master_url, worker_id): - """Send image batch to master via canonical JSON envelopes.""" - batch_size = image_batch.shape[0] - if batch_size == 0: - return - + """Send an image batch, optionally with audio, or an audio-only completion.""" encoded_audio = encode_audio_payload(audio) - session = await get_client_session() url = f"{master_url}/distributed/job_complete" - for batch_idx in range(batch_size): - img = tensor_to_pil(image_batch[batch_idx:batch_idx+1], 0) - byte_io = io.BytesIO() - img.save(byte_io, format='PNG', compress_level=0) - encoded_image = base64.b64encode(byte_io.getvalue()).decode('utf-8') - payload = { - "job_id": str(multi_job_id), - "worker_id": str(worker_id), - "batch_idx": int(batch_idx), - "image": f"data:image/png;base64,{encoded_image}", - "is_last": bool(batch_idx == batch_size - 1), - } - if payload["is_last"] and encoded_audio is not None: - payload["audio"] = encoded_audio + payloads = [] + batch_size = 0 if image_batch is None else image_batch.shape[0] + if batch_size == 0: + if encoded_audio is None: + raise ValueError("Worker completion requires image or audio data") + payloads.append( + { + "job_id": str(multi_job_id), + "worker_id": str(worker_id), + "batch_idx": 0, + "audio": encoded_audio, + "is_last": True, + } + ) + else: + for batch_idx in range(batch_size): + img = tensor_to_pil(image_batch[batch_idx:batch_idx+1], 0) + byte_io = io.BytesIO() + img.save(byte_io, format='PNG', compress_level=0) + encoded_image = base64.b64encode(byte_io.getvalue()).decode('utf-8') + payload = { + "job_id": str(multi_job_id), + "worker_id": str(worker_id), + "batch_idx": int(batch_idx), + "image": f"data:image/png;base64,{encoded_image}", + "is_last": bool(batch_idx == batch_size - 1), + } + if payload["is_last"] and encoded_audio is not None: + payload["audio"] = encoded_audio + payloads.append(payload) + + for payload in payloads: + timeout_seconds = 60 if "image" in payload else 600 try: async with session.post( url, json=payload, - timeout=aiohttp.ClientTimeout(total=60), + timeout=aiohttp.ClientTimeout(total=timeout_seconds), ) as response: response.raise_for_status() except Exception as e: - log(f"Worker - Failed to send canonical image envelope to master: {e}") + media_type = "image/audio" if "image" in payload else "audio-only" + log(f"Worker - Failed to send canonical {media_type} envelope to master: {e}") debug_log(f"Worker - Full error details: URL={url}") - raise # Re-raise to handle at caller level + raise def _combine_audio(self, master_audio, worker_audio, empty_audio, worker_order=None): """Combine audio from master and workers into a single audio output. @@ -257,8 +283,8 @@ class DistributedCollectorNode: images_on_cpu, delegate_mode: bool, fallback_images, - ) -> torch.Tensor: - """Assemble final tensor: master first, then workers in enabled order.""" + ): + """Assemble final tensor, or return None when the job contains only audio.""" ordered_tensors = [] if not delegate_mode and images_on_cpu is not None: for i in range(master_batch_size): @@ -289,15 +315,15 @@ class DistributedCollectorNode: if cpu_tensors: return ensure_contiguous(torch.cat(cpu_tensors, dim=0)) - elif fallback_images is not None: + if fallback_images is not None: return ensure_contiguous(fallback_images) - else: - raise ValueError("No image data collected from master or workers") + return None async def execute(self, images, audio, load_balance=False, multi_job_id="", is_worker=False, master_url="", enabled_worker_ids="[]", worker_batch_size=1, worker_id="", delegate_only=False): if is_worker: # Worker mode: send images and audio to master in a single batch - debug_log(f"Worker - Job {multi_job_id} complete. Sending {images.shape[0]} image(s) to master") + image_count = 0 if images is None else images.shape[0] + debug_log(f"Worker - Job {multi_job_id} complete. Sending {image_count} image(s) to master") await self.send_batch_to_master(images, audio, multi_job_id, master_url, worker_id) return (images, audio if audio is not None else self.EMPTY_AUDIO) else: @@ -332,14 +358,15 @@ class DistributedCollectorNode: master_audio = None debug_log(f"Master - Job {multi_job_id}: Delegate-only mode enabled, collecting exclusively from {num_workers} workers") else: - images_on_cpu = images.cpu() - master_batch_size = images.shape[0] + if images is None: + images_on_cpu = None + master_batch_size = 0 + else: + images_on_cpu = ensure_contiguous(images.cpu()) + master_batch_size = images.shape[0] master_audio = audio # Keep master's audio for later debug_log(f"Master - Job {multi_job_id}: Master has {master_batch_size} images, collecting from {num_workers} workers...") - # Ensure master images are contiguous - images_on_cpu = ensure_contiguous(images_on_cpu) - # Initialize storage for collected images and audio worker_images = {} # Dict to store images by worker_id and index @@ -511,18 +538,19 @@ class DistributedCollectorNode: if multi_job_id in prompt_server.distributed_pending_jobs: del prompt_server.distributed_pending_jobs[multi_job_id] + combined_audio = self._combine_audio(master_audio, worker_audio, self.EMPTY_AUDIO, enabled_workers) try: combined = self._reorder_and_combine_tensors( worker_images, enabled_workers, master_batch_size, images_on_cpu, delegate_mode, images ) - debug_log(f"Master - Job {multi_job_id} complete. Combined {combined.shape[0]} images total " - f"(master: {master_batch_size}, workers: {combined.shape[0] - master_batch_size})") - - # Combine audio from master and workers - combined_audio = self._combine_audio(master_audio, worker_audio, self.EMPTY_AUDIO, enabled_workers) + if combined is None: + debug_log(f"Master - Job {multi_job_id} complete with audio only") + else: + debug_log(f"Master - Job {multi_job_id} complete. Combined {combined.shape[0]} images total " + f"(master: {master_batch_size}, workers: {combined.shape[0] - master_batch_size})") return (combined, combined_audio) except Exception as e: log(f"Master - Error combining images: {e}") - # Return just the master images as fallback - return (images, audio if audio is not None else self.EMPTY_AUDIO) + # Preserve collected audio even when image assembly fails. + return (images, combined_audio) diff --git a/tests/api/test_distributed_queue.py b/tests/api/test_distributed_queue.py index 55694a3..ac6a333 100644 --- a/tests/api/test_distributed_queue.py +++ b/tests/api/test_distributed_queue.py @@ -302,6 +302,77 @@ class JobCompleteAudioPayloadTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(queued["audio"]["sample_rate"], 44100) self.assertEqual(tuple(queued["audio"]["waveform"].shape), (1, 2, 4)) + async def test_job_complete_accepts_audio_without_image(self): + queue = asyncio.Queue() + job_routes.prompt_server.distributed_jobs_lock = asyncio.Lock() + job_routes.prompt_server.distributed_pending_jobs = {"audio-only-job": queue} + request = _FakeRequest( + { + "job_id": "audio-only-job", + "worker_id": "worker-1", + "batch_idx": 0, + "audio": self._encoded_audio_payload(), + "is_last": True, + } + ) + + with patch.object(job_routes, "_decode_canonical_png_tensor") as decode_image: + response = await job_routes.job_complete_endpoint(request) + + self.assertEqual(response.status, 200) + decode_image.assert_not_called() + queued = await queue.get() + self.assertIsNone(queued["tensor"]) + self.assertEqual(queued["audio"]["sample_rate"], 44100) + + async def test_job_complete_rejects_payload_without_image_or_audio(self): + request = _FakeRequest( + { + "job_id": "job-1", + "worker_id": "worker-1", + "batch_idx": 0, + "is_last": True, + } + ) + + response = await job_routes.job_complete_endpoint(request) + + self.assertEqual(response.status, 400) + self.assertIn("image or audio", response.payload.get("message", "").lower()) + + async def test_job_complete_rejects_invalid_image_even_when_audio_is_present(self): + request = _FakeRequest( + { + "job_id": "job-1", + "worker_id": "worker-1", + "batch_idx": 0, + "image": 123, + "audio": self._encoded_audio_payload(), + "is_last": True, + } + ) + + response = await job_routes.job_complete_endpoint(request) + + self.assertEqual(response.status, 400) + self.assertIn("image", response.payload.get("message", "").lower()) + + async def test_job_complete_returns_400_for_malformed_audio(self): + request = _FakeRequest( + { + "job_id": "job-1", + "worker_id": "worker-1", + "batch_idx": 0, + "audio": {"data": "AAAA", "shape": [1, 2], "dtype": "float32"}, + "is_last": True, + } + ) + + response = await job_routes.job_complete_endpoint(request) + + self.assertEqual(response.status, 400) + self.assertIn("audio.shape", response.payload.get("message", "").lower()) + def test_decode_audio_payload_rejects_bad_shape(self): bad = { "sample_rate": 44100, diff --git a/tests/test_collector_list_inputs.py b/tests/test_collector_list_inputs.py index e8de4f1..d62312d 100644 --- a/tests/test_collector_list_inputs.py +++ b/tests/test_collector_list_inputs.py @@ -124,6 +124,49 @@ def test_collector_opts_into_comfyui_list_inputs(): assert collector.INPUT_IS_LIST is True +def test_collector_exposes_images_as_optional_input(): + input_types = _load_collector_module().DistributedCollectorNode.INPUT_TYPES() + + assert "images" not in input_types["required"] + assert input_types["optional"]["images"] == ("IMAGE",) + + +def test_audio_only_pass_through_returns_no_images_and_preserves_audio(): + collector = _load_collector_module().DistributedCollectorNode() + audio = {"waveform": torch.ones(1, 2, 4), "sample_rate": 48000} + + images, returned_audio = collector.run(images=None, audio=[audio]) + + assert images is None + assert returned_audio is audio + + +def test_collector_rejects_missing_images_and_audio(): + collector = _load_collector_module().DistributedCollectorNode() + + try: + collector.run(images=None, audio=None) + except ValueError as exc: + assert "image or audio" in str(exc).lower() + else: + raise AssertionError("Expected collector to reject a run with no media input") + + +def test_delegate_only_master_allows_no_local_media_input(): + collector = _load_collector_module().DistributedCollectorNode() + + images, audio = collector.run( + images=None, + audio=None, + multi_job_id=["delegate-audio-job"], + delegate_only=[True], + enabled_worker_ids=["[]"], + ) + + assert images is None + assert tuple(audio["waveform"].shape) == (1, 2, 1) + + def test_pass_through_collapses_comfyui_image_list_to_batch_and_unwraps_hidden_inputs(): collector = _load_collector_module().DistributedCollectorNode() first = torch.zeros(1, 2, 2, 3) @@ -203,3 +246,119 @@ def test_worker_list_input_sends_one_completion_sequence_with_last_only_on_final assert [payload["is_last"] for payload in posted_payloads] == [False, True] assert {payload["job_id"] for payload in posted_payloads} == {"job-list-1"} assert {payload["worker_id"] for payload in posted_payloads} == {"worker-a"} + + +def test_audio_only_worker_sends_one_completion_without_image(): + module = _load_collector_module() + collector = module.DistributedCollectorNode() + audio = {"waveform": torch.ones(1, 2, 4), "sample_rate": 48000} + posted = [] + + class _FakeResponse: + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + def raise_for_status(self): + return None + + class _FakeSession: + def post(self, url, json, timeout): + posted.append((url, json, timeout.total)) + return _FakeResponse() + + async def _fake_get_client_session(): + return _FakeSession() + + module.get_client_session = _fake_get_client_session + module.encode_audio_payload = lambda value: {"encoded": value is audio} + + images, returned_audio = collector.run( + images=None, + audio=[audio], + multi_job_id=["audio-job"], + is_worker=[True], + master_url=["http://master"], + worker_id=["worker-a"], + ) + + assert images is None + assert returned_audio is audio + assert len(posted) == 1 + assert posted[0][0] == "http://master/distributed/job_complete" + assert posted[0][2] == 600 + assert posted[0][1] == { + "job_id": "audio-job", + "worker_id": "worker-a", + "batch_idx": 0, + "audio": {"encoded": True}, + "is_last": True, + } + + +def test_audio_only_master_combines_local_and_worker_audio(): + module = _load_collector_module() + collector = module.DistributedCollectorNode() + master_audio = {"waveform": torch.ones(1, 2, 2), "sample_rate": 48000} + worker_audio = {"waveform": torch.full((1, 2, 3), 2.0), "sample_rate": 48000} + module.prompt_server.distributed_jobs_lock = asyncio.Lock() + queue = asyncio.Queue() + queue.put_nowait( + { + "worker_id": "worker-a", + "image_index": 0, + "tensor": None, + "audio": worker_audio, + "is_last": True, + } + ) + module.prompt_server.distributed_pending_jobs = {"audio-job": queue} + + images, combined_audio = asyncio.run( + collector.execute( + images=None, + audio=master_audio, + multi_job_id="audio-job", + enabled_worker_ids='["worker-a"]', + ) + ) + + assert images is None + assert combined_audio["sample_rate"] == 48000 + assert tuple(combined_audio["waveform"].shape) == (1, 2, 5) + assert torch.equal(combined_audio["waveform"][..., :2], master_audio["waveform"]) + assert torch.equal(combined_audio["waveform"][..., 2:], worker_audio["waveform"]) + + +def test_delegate_only_audio_collects_worker_audio_without_placeholder_image(): + module = _load_collector_module() + collector = module.DistributedCollectorNode() + worker_audio = {"waveform": torch.full((1, 2, 3), 2.0), "sample_rate": 48000} + module.prompt_server.distributed_jobs_lock = asyncio.Lock() + queue = asyncio.Queue() + queue.put_nowait( + { + "worker_id": "worker-a", + "image_index": 0, + "tensor": None, + "audio": worker_audio, + "is_last": True, + } + ) + module.prompt_server.distributed_pending_jobs = {"delegate-audio-job": queue} + + images, combined_audio = asyncio.run( + collector.execute( + images=None, + audio=None, + multi_job_id="delegate-audio-job", + enabled_worker_ids='["worker-a"]', + delegate_only=True, + ) + ) + + assert images is None + assert combined_audio["sample_rate"] == 48000 + assert torch.equal(combined_audio["waveform"], worker_audio["waveform"]) diff --git a/tests/test_prompt_transform.py b/tests/test_prompt_transform.py index 2bd200a..2ceca7c 100644 --- a/tests/test_prompt_transform.py +++ b/tests/test_prompt_transform.py @@ -71,6 +71,15 @@ def _collector_only_prompt(): } +def _audio_only_prompt(): + """1(Audio source) → 2(DistributedCollector) → 3(Audio sink).""" + return { + "1": {"class_type": "LoadAudio", "inputs": {}}, + "2": {"class_type": "DistributedCollector", "inputs": {"audio": ["1", 0]}}, + "3": {"class_type": "SaveAudio", "inputs": {"audio": ["2", 1]}}, + } + + def _delegate_prompt(): """1 → 2 → 3(DistributedCollector) → 4(SaveImage)""" return { @@ -217,6 +226,35 @@ class PrunePromptForWorkerTests(unittest.TestCase): self.assertEqual(len(preview_nodes), 1) self.assertEqual(preview_nodes[0]["inputs"]["images"], ["4", 0]) + def test_injects_preview_audio_for_audio_only_collector(self): + result = pt.prune_prompt_for_worker(_audio_only_prompt()) + preview_nodes = [n for n in result.values() if n.get("class_type") == "PreviewAudio"] + self.assertEqual(len(preview_nodes), 1) + self.assertEqual(preview_nodes[0]["inputs"]["audio"], ["2", 1]) + self.assertFalse(any(n.get("class_type") == "PreviewImage" for n in result.values())) + + def test_prefers_image_preview_when_collector_has_images_and_audio(self): + prompt = _linear_prompt() + prompt["6"] = {"class_type": "LoadAudio", "inputs": {}} + prompt["4"]["inputs"]["audio"] = ["6", 0] + result = pt.prune_prompt_for_worker(prompt) + self.assertEqual( + len([n for n in result.values() if n.get("class_type") == "PreviewImage"]), + 1, + ) + self.assertFalse(any(n.get("class_type") == "PreviewAudio" for n in result.values())) + + def test_preserves_image_preview_for_distributed_upscale(self): + prompt = { + "1": {"class_type": "LoadImage", "inputs": {}}, + "2": {"class_type": "UltimateSDUpscaleDistributed", "inputs": {"upscaled_image": ["1", 0]}}, + "3": {"class_type": "SaveImage", "inputs": {"images": ["2", 0]}}, + } + result = pt.prune_prompt_for_worker(prompt) + preview_nodes = [n for n in result.values() if n.get("class_type") == "PreviewImage"] + self.assertEqual(len(preview_nodes), 1) + self.assertEqual(preview_nodes[0]["inputs"]["images"], ["2", 0]) + def test_no_preview_image_when_no_downstream(self): result = pt.prune_prompt_for_worker(_collector_only_prompt()) preview_nodes = [n for n in result.values() if n.get("class_type") == "PreviewImage"] @@ -242,7 +280,7 @@ class PrunePromptForWorkerTests(unittest.TestCase): def test_upscale_node_is_treated_as_distributed(self): prompt = { "1": {"class_type": "KSampler", "inputs": {}}, - "2": {"class_type": "UltimateSDUpscaleDistributed", "inputs": {"image": ["1", 0]}}, + "2": {"class_type": "UltimateSDUpscaleDistributed", "inputs": {"upscaled_image": ["1", 0]}}, "3": {"class_type": "SaveImage", "inputs": {"images": ["2", 0]}}, } result = pt.prune_prompt_for_worker(prompt) @@ -285,12 +323,22 @@ class PrepareDelegateMasterPromptTests(unittest.TestCase): placeholder_id = empty_nodes[0][0] self.assertEqual(result["3"]["inputs"]["images"], [placeholder_id, 0]) + def test_audio_only_collector_does_not_get_image_placeholder(self): + prompt = _audio_only_prompt() + result = pt.prepare_delegate_master_prompt(prompt, ["2"]) + empty_nodes = [n for n in result.values() if n.get("class_type") == "DistributedEmptyImage"] + self.assertEqual(empty_nodes, []) + self.assertNotIn("images", result["2"].get("inputs", {})) + self.assertNotIn("audio", result["2"].get("inputs", {})) + def test_one_placeholder_per_collector(self): """Two collectors → two placeholders.""" prompt = { - "1": {"class_type": "DistributedCollector", "inputs": {}}, - "2": {"class_type": "DistributedCollector", "inputs": {}}, + "1": {"class_type": "DistributedCollector", "inputs": {"images": ["10", 0]}}, + "2": {"class_type": "DistributedCollector", "inputs": {"images": ["11", 0]}}, "3": {"class_type": "SaveImage", "inputs": {"images": ["1", 0]}}, + "10": {"class_type": "LoadImage", "inputs": {}}, + "11": {"class_type": "LoadImage", "inputs": {}}, } result = pt.prepare_delegate_master_prompt(prompt, ["1", "2"]) empty_nodes = [n for n in result.values() if n.get("class_type") == "DistributedEmptyImage"]