From 189165f2ec948a7d52b26e271a76afb00d089abc Mon Sep 17 00:00:00 2001 From: robertvoy Date: Tue, 15 Jul 2025 19:19:00 +1000 Subject: [PATCH] cross-platform paths --- distributed.py | 18 +++++++++++ web/executionUtils.js | 69 +++++++++++++++++++++++++++++++++++++++---- web/workerUtils.js | 50 +++++++++++++++++++++++++++++++ 3 files changed, 131 insertions(+), 6 deletions(-) diff --git a/distributed.py b/distributed.py index 5dc0b1e..2b4eaca 100644 --- a/distributed.py +++ b/distributed.py @@ -1295,6 +1295,24 @@ async def load_image_endpoint(request): except Exception as e: return await handle_api_error(request, e, 500) +@server.PromptServer.instance.routes.get("/distributed/system_info") +async def system_info_endpoint(request): + """Return system information including platform details""" + import platform + + return web.json_response({ + "platform": { + "os_name": os.name, # 'nt' for Windows, 'posix' for Unix/Linux + "path_separator": os.sep, # '\\' on Windows, '/' on Linux + "system": platform.system(), # 'Windows', 'Linux', 'Darwin' + "machine": platform.machine(), # 'x86_64', 'AMD64', etc. + }, + "comfyui": { + # Can add ComfyUI-specific info here + "version": "distributed-extension", + } + }) + @server.PromptServer.instance.routes.post("/distributed/job_complete") async def job_complete_endpoint(request): try: diff --git a/web/executionUtils.js b/web/executionUtils.js index 9960de4..f7a40f6 100644 --- a/web/executionUtils.js +++ b/web/executionUtils.js @@ -1,7 +1,42 @@ import { api } from "../../scripts/api.js"; -import { findNodesByClass, findImageReferences, hasUpstreamNode, pruneWorkflowForWorker } from './workerUtils.js'; +import { findNodesByClass, findImageReferences, hasUpstreamNode, pruneWorkflowForWorker, getCachedWorkerSystemInfo } from './workerUtils.js'; import { TIMEOUTS } from './constants.js'; +/** + * Convert paths in the API prompt to match the target platform's separator + * @param {Object} apiPrompt - The workflow API prompt + * @param {string} targetSeparator - The target path separator ('\\' or '/') + * @returns {Object} The converted API prompt + */ +function convertPathsForPlatform(apiPrompt, targetSeparator) { + // Regex to identify likely file paths with extensions + const isLikelyFilename = (value) => { + return value.match(/\.(ckpt|safetensors|pt|pth|bin|yaml|json|png|jpg|jpeg|webp|gif|bmp|latent|txt|vae|lora|embedding)(\s*\[\w+\])?$/i); + }; + + function convert(obj) { + if (typeof obj === 'string') { + // Only convert strings that look like file paths + if ((obj.includes('\\') || obj.includes('/')) && isLikelyFilename(obj)) { + // Replace any path separator with the target one + return obj.replace(/[\\\/]/g, targetSeparator); + } + return obj; + } else if (Array.isArray(obj)) { + return obj.map(convert); + } else if (typeof obj === 'object' && obj !== null) { + const newObj = {}; + for (const [key, value] of Object.entries(obj)) { + newObj[key] = convert(value); + } + return newObj; + } + return obj; + } + + return convert(apiPrompt); +} + export function setupInterceptor(extension) { api.queuePrompt = async (number, prompt) => { if (extension.isEnabled) { @@ -63,7 +98,7 @@ export async function executeParallelDistributed(extension, promptWrapper) { job_id_map: job_id_map // Pass the map of unique IDs }; - const jobApiPrompt = prepareApiPromptForParticipant( + const jobApiPrompt = await prepareApiPromptForParticipant( extension, promptWrapper.output, participantId, options ); @@ -97,7 +132,7 @@ export async function executeParallelDistributed(extension, promptWrapper) { } } -export function prepareApiPromptForParticipant(extension, baseApiPrompt, participantId, options = {}) { +export async function prepareApiPromptForParticipant(extension, baseApiPrompt, participantId, options = {}) { let jobApiPrompt = JSON.parse(JSON.stringify(baseApiPrompt)); const isMaster = participantId === 'master'; @@ -106,9 +141,31 @@ export function prepareApiPromptForParticipant(extension, baseApiPrompt, partici const upscaleNodes = findNodesByClass(jobApiPrompt, "UltimateSDUpscaleDistributed"); const allDistributedNodes = [...collectorNodes, ...upscaleNodes]; - // For workers, prune the workflow to only include distributed node dependencies - if (!isMaster && allDistributedNodes.length > 0) { - jobApiPrompt = pruneWorkflowForWorker(extension, jobApiPrompt, allDistributedNodes); + // For workers, handle platform-specific path conversion + if (!isMaster) { + const workerInfo = extension.config.workers.find(w => w.id === participantId); + + if (workerInfo && workerInfo.host) { + // Remote or cloud worker - needs path translation + try { + const workerUrl = extension.getWorkerUrl(workerInfo); + const systemInfo = await getCachedWorkerSystemInfo(workerUrl); + const targetSeparator = systemInfo.platform.path_separator; + + // Convert paths to match worker's platform + jobApiPrompt = convertPathsForPlatform(jobApiPrompt, targetSeparator); + + extension.log(`Converted paths for ${systemInfo.platform.system} worker ${participantId} (separator: '${targetSeparator}')`, "debug"); + } catch (e) { + extension.log(`Failed to get system info for worker ${participantId}: ${e.message}`, "warn"); + // Continue without path conversion + } + } + + // Prune the workflow to only include distributed node dependencies + if (allDistributedNodes.length > 0) { + jobApiPrompt = pruneWorkflowForWorker(extension, jobApiPrompt, allDistributedNodes); + } } // Handle image references for remote workers diff --git a/web/workerUtils.js b/web/workerUtils.js index 745758a..63280f7 100644 --- a/web/workerUtils.js +++ b/web/workerUtils.js @@ -263,4 +263,54 @@ export function hasUpstreamNode(apiPrompt, nodeId, upstreamType) { } return false; +} + +/** + * Get system information from a worker + * @param {string} workerUrl - The worker URL + * @returns {Promise} System information including platform details + */ +export async function getWorkerSystemInfo(workerUrl) { + try { + const response = await fetch(`${workerUrl}/distributed/system_info`); + if (!response.ok) { + throw new Error(`HTTP ${response.status}`); + } + return await response.json(); + } catch (error) { + console.warn(`Failed to get system info from ${workerUrl}:`, error); + // Return sensible defaults + return { + platform: { + os_name: 'posix', // Assume Linux + path_separator: '/', + system: 'Linux' + } + }; + } +} + +// Cache system info to avoid repeated calls +const systemInfoCache = new Map(); + +/** + * Get cached system information from a worker + * @param {string} workerUrl - The worker URL + * @returns {Promise} Cached or fresh system information + */ +export async function getCachedWorkerSystemInfo(workerUrl) { + if (systemInfoCache.has(workerUrl)) { + return systemInfoCache.get(workerUrl); + } + + const info = await getWorkerSystemInfo(workerUrl); + systemInfoCache.set(workerUrl, info); + return info; +} + +/** + * Clear the system info cache + */ +export function clearSystemInfoCache() { + systemInfoCache.clear(); } \ No newline at end of file