add process detection for TCP port usage and improve native mode backend checks in launch_app function
This commit is contained in:
Ross
@@ -35,6 +35,9 @@ from fastapi.responses import RedirectResponse
|
||||
from starlette.middleware.base import BaseHTTPMiddleware
|
||||
import typer
|
||||
import logging
|
||||
import subprocess
|
||||
import platform
|
||||
import importlib.util
|
||||
|
||||
# traceback was previously imported for debugging but is unused
|
||||
|
||||
@@ -732,6 +735,102 @@ def launch_app(
|
||||
pass
|
||||
except pynng.exceptions.AddressInUse:
|
||||
logger.debug("Address in use")
|
||||
# Try to detect which process is holding the TCP port (helpful on Windows)
|
||||
try:
|
||||
port_str = SOCKET_PATH.split(':')[-1]
|
||||
port = int(port_str)
|
||||
except Exception:
|
||||
port = None
|
||||
|
||||
if port:
|
||||
def find_process_using_port(port: int):
|
||||
"""Return a list of (pid, name) tuples for processes listening on TCP port.
|
||||
|
||||
Uses psutil if available; otherwise falls back to parsing `netstat -ano` and
|
||||
resolving PIDs with `tasklist` on Windows. Returns empty list if none found.
|
||||
"""
|
||||
results = []
|
||||
# Prefer psutil (fast, cross-platform) when available
|
||||
try:
|
||||
import psutil
|
||||
|
||||
for conn in psutil.net_connections(kind='tcp'):
|
||||
laddr = conn.laddr
|
||||
if not laddr:
|
||||
continue
|
||||
# laddr can be an address tuple (ip, port)
|
||||
try:
|
||||
conn_port = laddr.port
|
||||
except Exception:
|
||||
# on some platforms laddr may be a string
|
||||
conn_port = int(str(laddr).split(':')[-1])
|
||||
if conn_port == port and conn.status in ('LISTEN', 'LISTENING'):
|
||||
pid = conn.pid
|
||||
name = None
|
||||
try:
|
||||
if pid:
|
||||
p = psutil.Process(pid)
|
||||
name = p.name()
|
||||
except Exception:
|
||||
name = None
|
||||
results.append((pid, name))
|
||||
return results
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Fallback: use netstat (Windows) and tasklist to resolve PIDs
|
||||
try:
|
||||
if platform.system().lower().startswith('win'):
|
||||
cmd = ['netstat', '-ano', '-p', 'tcp']
|
||||
proc = subprocess.run(cmd, capture_output=True, text=True)
|
||||
out = proc.stdout.splitlines()
|
||||
pids = set()
|
||||
for line in out:
|
||||
parts = line.split()
|
||||
# Expected format: Proto Local Address Foreign Address State PID
|
||||
if len(parts) >= 5:
|
||||
local = parts[1]
|
||||
state = parts[3]
|
||||
pid = parts[4]
|
||||
# local can be like '0.0.0.0:9976' or '[::]:9976'
|
||||
if ':' in local and state.upper() in ('LISTEN', 'LISTENING'):
|
||||
try:
|
||||
local_port = int(local.rsplit(':', 1)[1])
|
||||
except Exception:
|
||||
continue
|
||||
if local_port == port:
|
||||
try:
|
||||
pid_i = int(pid)
|
||||
pids.add(pid_i)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
for pid in pids:
|
||||
# Resolve process name via tasklist
|
||||
try:
|
||||
tl = subprocess.run(['tasklist', '/FI', f'PID eq {pid}', '/FO', 'CSV', '/NH'], capture_output=True, text=True)
|
||||
line = tl.stdout.strip()
|
||||
if line and '"' in line:
|
||||
# CSV like "process.exe","1234","..."
|
||||
cols = [c.strip('"') for c in line.split(',')]
|
||||
name = cols[0] if cols else None
|
||||
else:
|
||||
name = None
|
||||
except Exception:
|
||||
name = None
|
||||
results.append((pid, name))
|
||||
return results
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return results
|
||||
|
||||
holders = find_process_using_port(port)
|
||||
if holders:
|
||||
for pid, name in holders:
|
||||
logger.error(f"Port {port} appears in use by PID={pid} name={name}")
|
||||
else:
|
||||
logger.debug(f"No process detected listening on port {port} via psutil/netstat")
|
||||
try:
|
||||
with pynng.Pair0(dial=SOCKET_PATH, send_timeout=4000, recv_timeout=4000) as sub:
|
||||
sub.send(b"loaded")
|
||||
@@ -755,40 +854,25 @@ def launch_app(
|
||||
logger.debug(f"Run on port {port}")
|
||||
|
||||
# If native mode was requested, ensure pywebview has a usable GUI
|
||||
# backend (Qt or GTK). If not present, disable native mode to avoid
|
||||
# runtime WebViewException errors when webview.initialize() runs.
|
||||
# backend (Qt or GTK). Use importlib.util.find_spec to check for
|
||||
# available backends without importing them (avoids side-effects).
|
||||
if native_mode:
|
||||
try:
|
||||
import webview
|
||||
|
||||
backend_available = False
|
||||
# Check common backends: PyQt5/PySide6 or GTK (gi)
|
||||
try:
|
||||
import PyQt5 # type: ignore
|
||||
backend_available = True
|
||||
except Exception:
|
||||
try:
|
||||
import PySide6 # type: ignore
|
||||
backend_available = True
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if not backend_available:
|
||||
try:
|
||||
# GTK
|
||||
from gi.repository import Gtk # type: ignore
|
||||
|
||||
backend_available = True
|
||||
except Exception:
|
||||
backend_available = False
|
||||
|
||||
if not backend_available:
|
||||
logger.debug(
|
||||
"pywebview backend (Qt/GTK) not available; disabling native mode"
|
||||
)
|
||||
if importlib.util.find_spec('webview') is None:
|
||||
logger.debug('pywebview package not found; disabling native mode')
|
||||
native_mode = False
|
||||
else:
|
||||
backend_available = False
|
||||
if importlib.util.find_spec('PyQt5') or importlib.util.find_spec('PySide6'):
|
||||
backend_available = True
|
||||
elif importlib.util.find_spec('gi'):
|
||||
backend_available = True
|
||||
|
||||
if not backend_available:
|
||||
logger.debug('pywebview backend (Qt/GTK) not available; disabling native mode')
|
||||
native_mode = False
|
||||
except Exception as e:
|
||||
logger.debug(f"pywebview import/check failed: {e}; disabling native mode")
|
||||
logger.debug(f'pywebview import/check failed: {e}; disabling native mode')
|
||||
native_mode = False
|
||||
|
||||
# Try to run with the requested mode. If native startup fails at
|
||||
|
||||
Reference in New Issue
Block a user