""" commands/utility.py =================== Quality-of-life utility commands for the Discord ComfyUI bot. Commands provided: - ping: Show bot latency (Discord WebSocket round-trip). - status: Full overview of bot health, ComfyUI connectivity, workflow state, and queue. - queue-status: Quick view of pending job count and worker state. - uptime: How long the bot has been running since it connected. """ from __future__ import annotations import logging from datetime import datetime, timezone from discord.ext import commands logger = logging.getLogger(__name__) def _format_uptime(start_time: datetime) -> str: """Return a human-readable uptime string from a UTC start datetime.""" delta = datetime.now(timezone.utc) - start_time total_seconds = int(delta.total_seconds()) days, remainder = divmod(total_seconds, 86400) hours, remainder = divmod(remainder, 3600) minutes, seconds = divmod(remainder, 60) if days: return f"{days}d {hours}h {minutes}m {seconds}s" if hours: return f"{hours}h {minutes}m {seconds}s" return f"{minutes}m {seconds}s" def setup_utility_commands(bot): """ Register quality-of-life utility commands with the bot. Parameters ---------- bot : commands.Bot The Discord bot instance. """ @bot.command(name="ping", extras={"category": "Utility"}) async def ping_command(ctx: commands.Context) -> None: """ Show the bot's current Discord WebSocket latency. Usage: ttr!ping """ latency_ms = round(bot.latency * 1000) await ctx.reply(f"Pong! Latency: **{latency_ms} ms**", mention_author=False) @bot.command(name="status", extras={"category": "Utility"}) async def status_command(ctx: commands.Context) -> None: """ Show a full health overview of the bot and ComfyUI. Displays: - Bot latency and uptime - ComfyUI server address and reachability - Whether a workflow template is loaded - Current workflow changes (prompt / negative_prompt / input_image) - Job queue size and worker state Usage: ttr!status """ latency_ms = round(bot.latency * 1000) # Uptime if hasattr(bot, "start_time") and bot.start_time: uptime_str = _format_uptime(bot.start_time) else: uptime_str = "N/A" # ComfyUI info comfy_ok = hasattr(bot, "comfy") and bot.comfy is not None comfy_server = bot.comfy.server_address if comfy_ok else "not configured" comfy_reachable = await bot.comfy.check_connection() if comfy_ok else False workflow_loaded = comfy_ok and bot.comfy.get_workflow_template() is not None # ComfyUI queue comfy_pending = 0 comfy_running = 0 if comfy_ok: q = await bot.comfy.get_comfy_queue() if q: comfy_pending = len(q.get("queue_pending", [])) comfy_running = len(q.get("queue_running", [])) # Workflow state summary changes_parts: list[str] = [] if comfy_ok: overrides = bot.comfy.state_manager.get_overrides() if overrides.get("prompt"): changes_parts.append("prompt") if overrides.get("negative_prompt"): changes_parts.append("negative_prompt") if overrides.get("input_image"): changes_parts.append(f"input_image: {overrides['input_image']}") if overrides.get("seed") is not None: changes_parts.append(f"seed={overrides['seed']}") changes_summary = ", ".join(changes_parts) if changes_parts else "none" conn_status = ( "reachable" if comfy_reachable else ("unreachable" if comfy_ok else "not configured") ) lines = [ "**Bot**", f" Latency : {latency_ms} ms", f" Uptime : {uptime_str}", "", f"**ComfyUI** — `{comfy_server}`", f" Connection : {conn_status}", f" Queue : {comfy_running} running, {comfy_pending} pending", f" Workflow : {'loaded' if workflow_loaded else 'not loaded'}", f" Changes set : {changes_summary}", ] await ctx.reply("\n".join(lines), mention_author=False) @bot.command(name="queue-status", aliases=["qs", "qstatus"], extras={"category": "Utility"}) async def queue_status_command(ctx: commands.Context) -> None: """ Show the current ComfyUI queue depth. Usage: ttr!queue-status ttr!qs """ if not hasattr(bot, "comfy") or not bot.comfy: await ctx.reply("ComfyUI client is not configured.", mention_author=False) return q = await bot.comfy.get_comfy_queue() if q is None: await ctx.reply("Could not reach ComfyUI server.", mention_author=False) return pending = len(q.get("queue_pending", [])) running = len(q.get("queue_running", [])) await ctx.reply( f"ComfyUI queue: **{running}** running, **{pending}** pending.", mention_author=False, ) @bot.command(name="uptime", extras={"category": "Utility"}) async def uptime_command(ctx: commands.Context) -> None: """ Show how long the bot has been running since it last connected. Usage: ttr!uptime """ if not hasattr(bot, "start_time") or not bot.start_time: await ctx.reply("Uptime information is not available.", mention_author=False) return uptime_str = _format_uptime(bot.start_time) await ctx.reply(f"Uptime: **{uptime_str}**", mention_author=False) @bot.command(name="comfy-stats", aliases=["cstats"], extras={"category": "Utility"}) async def comfy_stats_command(ctx: commands.Context) -> None: """ Show GPU and system stats from the ComfyUI server. Displays OS, Python version, and per-device VRAM usage reported by the ComfyUI ``/system_stats`` endpoint. Usage: ttr!comfy-stats ttr!cstats """ if not hasattr(bot, "comfy") or not bot.comfy: await ctx.reply("ComfyUI client is not configured.", mention_author=False) return stats = await bot.comfy.get_system_stats() if stats is None: await ctx.reply( "Could not reach the ComfyUI server to fetch stats.", mention_author=False ) return system = stats.get("system", {}) devices = stats.get("devices", []) lines = [ f"**ComfyUI System Stats** — `{bot.comfy.server_address}`", f" OS : {system.get('os', 'N/A')}", f" Python : {system.get('python_version', 'N/A')}", ] if devices: lines.append("") lines.append("**Devices**") for dev in devices: name = dev.get("name", "unknown") vram_total = dev.get("vram_total", 0) vram_free = dev.get("vram_free", 0) vram_used = vram_total - vram_free def _mb(b: int) -> str: return f"{b / 1024 / 1024:.0f} MB" lines.append( f" {name} — {_mb(vram_used)} / {_mb(vram_total)} VRAM used" ) else: lines.append(" No device info available.") await ctx.reply("\n".join(lines), mention_author=False) @bot.command(name="comfy-queue", aliases=["cqueue", "cq"], extras={"category": "Utility"}) async def comfy_queue_command(ctx: commands.Context) -> None: """ Show the ComfyUI server's internal queue state. Displays jobs currently running and pending on the ComfyUI server itself (separate from the Discord bot's own job queue). Usage: ttr!comfy-queue ttr!cq """ if not hasattr(bot, "comfy") or not bot.comfy: await ctx.reply("ComfyUI client is not configured.", mention_author=False) return queue_data = await bot.comfy.get_comfy_queue() if queue_data is None: await ctx.reply( "Could not reach the ComfyUI server to fetch queue info.", mention_author=False ) return running = queue_data.get("queue_running", []) pending = queue_data.get("queue_pending", []) lines = [ f"**ComfyUI Server Queue** — `{bot.comfy.server_address}`", f" Running : {len(running)} job(s)", f" Pending : {len(pending)} job(s)", ] if running: lines.append("") lines.append("**Currently running**") for entry in running[:5]: # cap at 5 to avoid huge messages prompt_id = entry[1] if len(entry) > 1 else "unknown" lines.append(f" `{prompt_id}`") if pending: lines.append("") lines.append(f"**Pending** (showing up to 5 of {len(pending)})") for entry in pending[:5]: prompt_id = entry[1] if len(entry) > 1 else "unknown" lines.append(f" `{prompt_id}`") await ctx.reply("\n".join(lines), mention_author=False)