"""
Process Tracker Utility
Tracks which processes are currently alive for log file status indication.
Used by log viewer to distinguish logs from running vs terminated processes.
"""
import logging
import re
from pathlib import Path
from typing import Iterable, Set, Optional
from dataclasses import dataclass
logger = logging.getLogger(__name__)
[docs]
@dataclass
class ProcessInfo:
"""Information about a tracked process."""
pid: int
name: str
status: str
create_time: float
[docs]
class ProcessTracker:
"""
Tracks which processes are currently alive.
Used to determine if log files correspond to running or terminated processes.
"""
[docs]
def __init__(self):
"""Initialize process tracker."""
self.alive_pids: Set[int] = set()
self.process_info: dict[int, ProcessInfo] = {}
self._psutil_available = False
# Check if psutil is available
try:
import psutil
self._psutil_available = True
logger.debug("ProcessTracker initialized with psutil support")
except ImportError:
logger.warning("psutil not available - process tracking disabled")
[docs]
def update(self, target_pids: Iterable[int] | None = None) -> bool:
"""Update tracked process state.
Args:
target_pids: Optional PID set to check directly. When provided,
avoids scanning every process on the machine.
Returns:
True when tracked alive/dead state changed.
"""
if not self._psutil_available:
return False
try:
import psutil
old_alive_pids = self.alive_pids
target_pid_set = set(target_pids) if target_pids is not None else None
new_alive_pids = set()
new_process_info = {}
if target_pid_set is None:
process_iter = psutil.process_iter(['pid', 'name', 'status', 'create_time'])
for proc in process_iter:
try:
pid = proc.info['pid']
new_alive_pids.add(pid)
new_process_info[pid] = ProcessInfo(
pid=pid,
name=proc.info.get('name', 'unknown'),
status=proc.info.get('status', 'unknown'),
create_time=proc.info.get('create_time', 0),
)
except (psutil.NoSuchProcess, psutil.AccessDenied):
pass
else:
for pid in target_pid_set:
try:
proc = psutil.Process(pid)
new_alive_pids.add(pid)
new_process_info[pid] = ProcessInfo(
pid=pid,
name=proc.name(),
status=proc.status(),
create_time=proc.create_time(),
)
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
pass
self.alive_pids = new_alive_pids
self.process_info = new_process_info
changed = old_alive_pids != new_alive_pids
logger.debug(f"Updated process tracker: {len(self.alive_pids)} alive processes")
return changed
except Exception as e:
logger.warning(f"Failed to update process tracker: {e}")
return False
[docs]
def is_alive(self, pid: Optional[int]) -> bool:
"""
Check if a PID is currently alive.
Args:
pid: Process ID to check (None returns False)
Returns:
bool: True if process is alive, False otherwise
"""
if pid is None or not self._psutil_available:
return False
return pid in self.alive_pids
[docs]
def get_process_info(self, pid: int) -> Optional[ProcessInfo]:
"""
Get information about a process.
Args:
pid: Process ID
Returns:
ProcessInfo if process is alive, None otherwise
"""
return self.process_info.get(pid)
[docs]
def get_status_icon(self, pid: Optional[int]) -> str:
"""
Get status icon for a process.
Args:
pid: Process ID (None returns unknown icon)
Returns:
str: Status icon (🟢 alive, ⚫ dead, ❓ unknown)
"""
if pid is None or not self._psutil_available:
return "❓"
return "🟢" if self.is_alive(pid) else "⚫"
[docs]
def get_status_text(self, pid: Optional[int]) -> str:
"""
Get human-readable status text for a process.
Args:
pid: Process ID (None returns "Unknown")
Returns:
str: Status text ("Running", "Terminated", "Unknown")
"""
if pid is None or not self._psutil_available:
return "Unknown"
return "Running" if self.is_alive(pid) else "Terminated"
[docs]
def get_log_display_name(log_path: Path, process_tracker: ProcessTracker) -> str:
"""
Get display name for log file with process status indicator.
Args:
log_path: Path to log file
process_tracker: ProcessTracker instance
Returns:
str: Display name with status icon (e.g., "🟢 worker_12345.log")
"""
pid = extract_pid_from_log_filename(log_path)
icon = process_tracker.get_status_icon(pid)
return f"{icon} {log_path.name}"