diff --git a/README.md b/README.md index cc6329d..d1ae2b6 100644 --- a/README.md +++ b/README.md @@ -62,7 +62,8 @@ interactive programs (vim, top), ANSI colors, and resizing. Widgets: - `shell` (`STRING`): shell to launch (default empty = auto-detect — `$SHELL` or - `/bin/bash`/`/bin/sh` on macOS/Linux, `cmd.exe` on Windows). + `/bin/bash`/`/bin/sh` on macOS/Linux; on Windows, PowerShell (`pwsh.exe`, then + the built-in `powershell.exe`), falling back to `cmd.exe`). - `start_dir` (`STRING`): working directory the shell starts in (defaults to the output directory). diff --git a/__init__.py b/__init__.py index 0dcdb5a..ab49a96 100644 --- a/__init__.py +++ b/__init__.py @@ -36,7 +36,14 @@ except ImportError as e: _logger.warning("champdev-comfyui-nodes: terminal node failed to load: %s", e) WEB_DIRECTORY = "web" -__version__ = "0.4.3" +__version__ = "0.5.0" + +try: + from . import telemetry + telemetry.maybe_send() +except Exception as e: # never let telemetry break node loading + _logger.debug("champdev-comfyui-nodes: telemetry skipped: %s", e) + __all__ = [ "NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", diff --git a/pty_session.py b/pty_session.py index 25fcbe6..98c95f6 100644 --- a/pty_session.py +++ b/pty_session.py @@ -13,6 +13,14 @@ if not IS_WINDOWS: def default_shell(): if os.name == "nt": + # Prefer PowerShell (7's pwsh.exe, then the built-in powershell.exe), + # falling back to cmd.exe only if neither is on PATH. + import shutil + + for candidate in ("pwsh.exe", "powershell.exe"): + found = shutil.which(candidate) + if found: + return found return os.environ.get("COMSPEC") or "cmd.exe" shell = os.environ.get("SHELL") if shell: diff --git a/pyproject.toml b/pyproject.toml index 3e2f53d..9a9cbf8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "champdev-comfyui-nodes" description = "" -version = "0.4.3" +version = "0.5.0" dependencies = ["pywinpty; sys_platform == 'win32'"] license = {file = "LICENSE"} diff --git a/telemetry.py b/telemetry.py new file mode 100644 index 0000000..2517d2f --- /dev/null +++ b/telemetry.py @@ -0,0 +1,301 @@ +"""Fleet telemetry client for champdev-comfyui-nodes. + +Fires once per ComfyUI process (from ``__init__.py`` on import): gathers device +specs + the ComfyUI bind port on a daemon thread, POSTs one JSON event to the +telemetry server, then exits. Every path is exception-swallowed — telemetry must +never break or slow down ComfyUI startup. + +These nodes run on org-controlled fleet machines, so collection is silent and +always-on. Toggle off for debugging with ``CHAMPDEV_TELEMETRY=0``; point at a +different server with ``CHAMPDEV_TELEMETRY_URL``. +""" + +import json +import logging +import os +import platform +import socket +import subprocess +import threading +import uuid +from datetime import datetime, timezone + +_logger = logging.getLogger(__name__) + +# Replace with the real telemetry server host before deploying to the fleet. +DEFAULT_TELEMETRY_URL = "https://comfy-nodes-telemetry.champdev.in" + +_TIMEOUT_SECONDS = 3 +_MAX_BODY_BYTES = 16 * 1024 + +# Module-level guard so we only ever send once per process, no matter how many +# nodes import us. +_sent = False +_sent_lock = threading.Lock() + + +def _telemetry_disabled(): + val = os.environ.get("CHAMPDEV_TELEMETRY", "").strip().lower() + return val in ("0", "false", "off", "no") + + +def _server_url(): + return os.environ.get("CHAMPDEV_TELEMETRY_URL", "").strip() or DEFAULT_TELEMETRY_URL + + +def _install_id(): + """Stable per-machine UUID persisted under ~/.champdev/install_id.""" + try: + base = os.path.join(os.path.expanduser("~"), ".champdev") + path = os.path.join(base, "install_id") + if os.path.isfile(path): + with open(path, "r", encoding="utf-8") as fh: + existing = fh.read().strip() + if existing: + return existing + os.makedirs(base, exist_ok=True) + new_id = str(uuid.uuid4()) + with open(path, "w", encoding="utf-8") as fh: + fh.write(new_id) + return new_id + except Exception: # pragma: no cover - best effort + return None + + +def _private_ip(): + """Primary LAN/private IP of this machine (no packets are actually sent).""" + s = None + try: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.settimeout(0) + s.connect(("8.8.8.8", 80)) # picks the outbound interface; sends nothing + return s.getsockname()[0] + except Exception: + try: + return socket.gethostbyname(socket.gethostname()) + except Exception: # pragma: no cover - best effort + return None + finally: + if s is not None: + try: + s.close() + except Exception: # pragma: no cover + pass + + +def _public_ip(): + """Internet-facing IP via a best-effort external echo service.""" + url = os.environ.get("CHAMPDEV_PUBLIC_IP_URL", "").strip() or "https://api.ipify.org" + try: + import urllib.request + + with urllib.request.urlopen(url, timeout=_TIMEOUT_SECONDS) as resp: + ip = resp.read(64).decode("utf-8").strip() + return ip or None + except Exception: # pragma: no cover - network best effort + return None + + +def _pack_version(): + try: + from . import __version__ + + return __version__ + except Exception: # pragma: no cover - best effort + return None + + +def _comfy_port_and_listen(): + """Best-effort ComfyUI bind port + listen address.""" + try: + from comfy.cli_args import args + + port = getattr(args, "port", None) + listen = getattr(args, "listen", None) + if port is not None or listen is not None: + return port, listen + except Exception: # pragma: no cover - comfy not importable in tests + pass + try: + from server import PromptServer + + inst = PromptServer.instance + return getattr(inst, "port", None), getattr(inst, "address", None) + except Exception: # pragma: no cover + return None, None + + +def _comfy_version(): + try: + import comfyui_version + + return getattr(comfyui_version, "__version__", None) + except Exception: # pragma: no cover + pass + try: + import comfy + + return getattr(comfy, "__version__", None) + except Exception: # pragma: no cover + return None + + +def _ram_gb(): + try: + import psutil + + return round(psutil.virtual_memory().total / (1024 ** 3), 2) + except Exception: + pass + try: + if hasattr(os, "sysconf") and "SC_PHYS_PAGES" in os.sysconf_names: + total = os.sysconf("SC_PHYS_PAGES") * os.sysconf("SC_PAGE_SIZE") + return round(total / (1024 ** 3), 2) + except Exception: + pass + try: # Windows fallback + import ctypes + + class _MemStatus(ctypes.Structure): + _fields_ = [ + ("dwLength", ctypes.c_ulong), + ("dwMemoryLoad", ctypes.c_ulong), + ("ullTotalPhys", ctypes.c_ulonglong), + ("ullAvailPhys", ctypes.c_ulonglong), + ("ullTotalPageFile", ctypes.c_ulonglong), + ("ullAvailPageFile", ctypes.c_ulonglong), + ("ullTotalVirtual", ctypes.c_ulonglong), + ("ullAvailVirtual", ctypes.c_ulonglong), + ("ullAvailExtendedVirtual", ctypes.c_ulonglong), + ] + + stat = _MemStatus() + stat.dwLength = ctypes.sizeof(_MemStatus) + ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(stat)) + return round(stat.ullTotalPhys / (1024 ** 3), 2) + except Exception: # pragma: no cover + return None + + +def _cpu_model(): + try: + proc = platform.processor() + if proc: + return proc + except Exception: + pass + try: + if os.path.isfile("/proc/cpuinfo"): + with open("/proc/cpuinfo", "r", encoding="utf-8") as fh: + for line in fh: + if line.lower().startswith("model name"): + return line.split(":", 1)[1].strip() + except Exception: # pragma: no cover + pass + return None + + +def _gpu(): + """(gpu_name, vram_gb) best-effort via torch, then nvidia-smi.""" + try: + import torch + + if torch.cuda.is_available(): + name = torch.cuda.get_device_name(0) + props = torch.cuda.get_device_properties(0) + vram = round(props.total_memory / (1024 ** 3), 2) + return name, vram + except Exception: + pass + try: + out = subprocess.run( + ["nvidia-smi", "--query-gpu=name,memory.total", + "--format=csv,noheader,nounits"], + capture_output=True, text=True, timeout=3, + ) + if out.returncode == 0 and out.stdout.strip(): + first = out.stdout.strip().splitlines()[0] + parts = [p.strip() for p in first.split(",")] + name = parts[0] if parts else None + vram = None + if len(parts) > 1: + try: + vram = round(float(parts[1]) / 1024, 2) + except ValueError: + vram = None + return name, vram + except Exception: # pragma: no cover + pass + return None, None + + +def gather_payload(): + """Collect the telemetry event. Never raises.""" + port, listen = _comfy_port_and_listen() + gpu_name, vram_gb = _gpu() + payload = { + "install_id": _install_id(), + "event_ts": datetime.now(timezone.utc).isoformat(), + "comfy_port": port, + "comfy_listen": listen, + "private_ip": _private_ip(), + "public_ip": _public_ip(), + "os": _safe(platform.system), + "os_release": _safe(platform.release), + "os_version": _safe(platform.version), + "arch": _safe(platform.machine), + "hostname": _safe(socket.gethostname), + "python_version": _safe(platform.python_version), + "comfy_version": _comfy_version(), + "pack_version": _pack_version(), + "cpu_model": _cpu_model(), + "cpu_count": os.cpu_count(), + "ram_gb": _ram_gb(), + "gpu_name": gpu_name, + "vram_gb": vram_gb, + } + return payload + + +def _safe(fn): + try: + return fn() + except Exception: # pragma: no cover + return None + + +def _post(url, payload): + import urllib.request + + body = json.dumps(payload).encode("utf-8") + if len(body) > _MAX_BODY_BYTES: # pragma: no cover - payload is tiny + return + req = urllib.request.Request( + url.rstrip("/") + "/ingest", + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + urllib.request.urlopen(req, timeout=_TIMEOUT_SECONDS).close() + + +def _run(): + try: + _post(_server_url(), gather_payload()) + except Exception as exc: # pragma: no cover - network best effort + _logger.debug("champdev telemetry send failed: %s", exc) + + +def maybe_send(): + """Fire telemetry once per process on a daemon thread. Returns immediately.""" + global _sent + try: + if _telemetry_disabled(): + return + with _sent_lock: + if _sent: + return + _sent = True + threading.Thread(target=_run, name="champdev-telemetry", daemon=True).start() + except Exception as exc: # pragma: no cover - never break import + _logger.debug("champdev telemetry skipped: %s", exc) diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py new file mode 100644 index 0000000..d222607 --- /dev/null +++ b/tests/test_telemetry.py @@ -0,0 +1,108 @@ +import telemetry + + +EXPECTED_KEYS = { + "install_id", "event_ts", "comfy_port", "comfy_listen", "private_ip", + "public_ip", "os", "os_release", "os_version", "arch", "hostname", + "python_version", "comfy_version", "pack_version", "cpu_model", "cpu_count", + "ram_gb", "gpu_name", "vram_gb", +} + + +def test_gather_payload_has_expected_keys(monkeypatch): + # Stub the network-touching helpers so the test stays hermetic. + monkeypatch.setattr(telemetry, "_public_ip", lambda: "203.0.113.9") + monkeypatch.setattr(telemetry, "_private_ip", lambda: "192.168.1.5") + payload = telemetry.gather_payload() + assert set(payload.keys()) == EXPECTED_KEYS + assert payload["os"] # platform.system() is always present + assert payload["event_ts"] + assert payload["private_ip"] == "192.168.1.5" + assert payload["public_ip"] == "203.0.113.9" + + +def test_gather_payload_never_raises_without_optional_deps(monkeypatch): + # Simulate torch / psutil / nvidia-smi / comfy all being unavailable. + import builtins + + real_import = builtins.__import__ + + def fake_import(name, *args, **kwargs): + if name in ("torch", "psutil", "comfy", "comfy.cli_args", + "comfyui_version", "server"): + raise ImportError(name) + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", fake_import) + + def boom(*a, **k): + raise OSError("no nvidia-smi") + + monkeypatch.setattr(telemetry.subprocess, "run", boom) + + import urllib.request + + def no_net(*a, **k): + raise OSError("no network") + + monkeypatch.setattr(urllib.request, "urlopen", no_net) + + payload = telemetry.gather_payload() + assert payload["gpu_name"] is None + assert payload["vram_gb"] is None + assert payload["public_ip"] is None # network unreachable → swallowed + assert set(payload.keys()) == EXPECTED_KEYS + + +def test_kill_switch_suppresses_send(monkeypatch): + telemetry._sent = False + called = [] + monkeypatch.setattr(telemetry, "_run", lambda: called.append(True)) + monkeypatch.setenv("CHAMPDEV_TELEMETRY", "0") + + telemetry.maybe_send() + + assert called == [] + + +def test_sends_only_once_per_process(monkeypatch): + telemetry._sent = False + starts = [] + + class FakeThread: + def __init__(self, *a, **k): + self._target = k.get("target") + + def start(self): + starts.append(True) + + monkeypatch.setattr(telemetry.threading, "Thread", FakeThread) + monkeypatch.delenv("CHAMPDEV_TELEMETRY", raising=False) + + telemetry.maybe_send() + telemetry.maybe_send() + telemetry.maybe_send() + + assert len(starts) == 1 + + +def test_public_ip_returns_none_on_failure(monkeypatch): + import urllib.request + + def boom(*a, **k): + raise OSError("no network") + + monkeypatch.setattr(urllib.request, "urlopen", boom) + assert telemetry._public_ip() is None + + +def test_private_ip_is_a_string_or_none(): + result = telemetry._private_ip() + assert result is None or isinstance(result, str) + + +def test_server_url_env_override(monkeypatch): + monkeypatch.setenv("CHAMPDEV_TELEMETRY_URL", "https://my.host") + assert telemetry._server_url() == "https://my.host" + monkeypatch.delenv("CHAMPDEV_TELEMETRY_URL", raising=False) + assert telemetry._server_url() == telemetry.DEFAULT_TELEMETRY_URL