mirror of
https://github.com/Omni-guides/Jackify.git
synced 2026-01-17 19:47:00 +01:00
Sync from development - prepare for v0.2.0
This commit is contained in:
Omni
318
jackify/shared/progress_models.py
Normal file
318
jackify/shared/progress_models.py
Normal file
@@ -0,0 +1,318 @@
|
||||
"""
|
||||
Progress Data Models
|
||||
|
||||
Shared data models for representing installation progress state.
|
||||
Used by both parser and GUI components.
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import List, Dict, Optional
|
||||
from enum import Enum
|
||||
import time
|
||||
|
||||
|
||||
class InstallationPhase(Enum):
|
||||
"""Installation phases that can be detected."""
|
||||
UNKNOWN = "unknown"
|
||||
INITIALIZATION = "initialization"
|
||||
DOWNLOAD = "download"
|
||||
EXTRACT = "extract"
|
||||
VALIDATE = "validate"
|
||||
INSTALL = "install"
|
||||
FINALIZE = "finalize"
|
||||
|
||||
|
||||
class OperationType(Enum):
|
||||
"""Types of operations being performed on files."""
|
||||
DOWNLOAD = "download"
|
||||
EXTRACT = "extract"
|
||||
VALIDATE = "validate"
|
||||
INSTALL = "install"
|
||||
UNKNOWN = "unknown"
|
||||
|
||||
|
||||
@dataclass
|
||||
class FileProgress:
|
||||
"""Represents progress for a single file operation."""
|
||||
filename: str
|
||||
operation: OperationType
|
||||
percent: float = 0.0 # 0-100
|
||||
current_size: int = 0 # Bytes processed
|
||||
total_size: int = 0 # Total bytes (0 if unknown)
|
||||
speed: float = -1.0 # Bytes per second (-1 = not provided by engine)
|
||||
last_update: float = field(default_factory=time.time)
|
||||
|
||||
def __post_init__(self):
|
||||
"""Ensure percent is in valid range."""
|
||||
self.percent = max(0.0, min(100.0, self.percent))
|
||||
|
||||
@property
|
||||
def is_complete(self) -> bool:
|
||||
"""Check if file operation is complete."""
|
||||
return self.percent >= 100.0 or (self.total_size > 0 and self.current_size >= self.total_size)
|
||||
|
||||
@property
|
||||
def size_display(self) -> str:
|
||||
"""Get human-readable size display."""
|
||||
if self.total_size > 0:
|
||||
return f"{self._format_bytes(self.current_size)}/{self._format_bytes(self.total_size)}"
|
||||
elif self.current_size > 0:
|
||||
return f"{self._format_bytes(self.current_size)}"
|
||||
else:
|
||||
return ""
|
||||
|
||||
@property
|
||||
def speed_display(self) -> str:
|
||||
"""Get human-readable speed display."""
|
||||
if self.speed <= 0:
|
||||
return ""
|
||||
return f"{self._format_bytes(int(self.speed))}/s"
|
||||
|
||||
@staticmethod
|
||||
def _format_bytes(bytes_val: int) -> str:
|
||||
"""Format bytes to human-readable format."""
|
||||
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
|
||||
if bytes_val < 1024.0:
|
||||
return f"{bytes_val:.1f}{unit}"
|
||||
bytes_val /= 1024.0
|
||||
return f"{bytes_val:.1f}PB"
|
||||
|
||||
|
||||
@dataclass
|
||||
class InstallationProgress:
|
||||
"""Complete installation progress state."""
|
||||
phase: InstallationPhase = InstallationPhase.UNKNOWN
|
||||
phase_name: str = "" # Human-readable phase name
|
||||
phase_step: int = 0 # Current step in phase
|
||||
phase_max_steps: int = 0 # Total steps in phase (0 if unknown)
|
||||
overall_percent: float = 0.0 # 0-100 overall progress
|
||||
data_processed: int = 0 # Bytes processed
|
||||
data_total: int = 0 # Total bytes (0 if unknown)
|
||||
active_files: List[FileProgress] = field(default_factory=list)
|
||||
speeds: Dict[str, float] = field(default_factory=dict) # Speed by operation type
|
||||
speed_timestamps: Dict[str, float] = field(default_factory=dict) # Last time each speed updated
|
||||
timestamp: float = field(default_factory=time.time)
|
||||
message: str = "" # Current status message
|
||||
texture_conversion_current: int = 0 # Current texture being converted
|
||||
texture_conversion_total: int = 0 # Total textures to convert
|
||||
bsa_building_current: int = 0 # Current BSA being built
|
||||
bsa_building_total: int = 0 # Total BSAs to build
|
||||
|
||||
def __post_init__(self):
|
||||
"""Ensure percent is in valid range."""
|
||||
self.overall_percent = max(0.0, min(100.0, self.overall_percent))
|
||||
|
||||
@property
|
||||
def phase_progress_text(self) -> str:
|
||||
"""Get phase progress text like '[12/14]'."""
|
||||
if self.phase_max_steps > 0:
|
||||
return f"[{self.phase_step}/{self.phase_max_steps}]"
|
||||
elif self.phase_step > 0:
|
||||
return f"[{self.phase_step}]"
|
||||
else:
|
||||
return ""
|
||||
|
||||
@property
|
||||
def data_progress_text(self) -> str:
|
||||
"""Get data progress text like '1.1GB/56.3GB'."""
|
||||
if self.data_total > 0:
|
||||
return f"{FileProgress._format_bytes(self.data_processed)}/{FileProgress._format_bytes(self.data_total)}"
|
||||
elif self.data_processed > 0:
|
||||
return f"{FileProgress._format_bytes(self.data_processed)}"
|
||||
else:
|
||||
return ""
|
||||
|
||||
def get_overall_speed_display(self) -> str:
|
||||
"""Get overall speed display from speeds dict or sum of active files."""
|
||||
def _fresh_speed(op_key: str) -> float:
|
||||
"""Return speed if recently updated, else 0."""
|
||||
if op_key not in self.speeds:
|
||||
return 0.0
|
||||
updated_at = self.speed_timestamps.get(op_key, 0.0)
|
||||
if updated_at == 0.0:
|
||||
return 0.0
|
||||
if time.time() - updated_at > 2.0:
|
||||
return 0.0
|
||||
return max(0.0, self.speeds.get(op_key, 0.0))
|
||||
|
||||
# Prefer aggregate speeds that match the current phase
|
||||
phase_operation_map = {
|
||||
InstallationPhase.DOWNLOAD: 'download',
|
||||
InstallationPhase.EXTRACT: 'extract',
|
||||
InstallationPhase.VALIDATE: 'validate',
|
||||
InstallationPhase.INSTALL: 'install',
|
||||
}
|
||||
active_op = phase_operation_map.get(self.phase)
|
||||
if active_op:
|
||||
op_speed = _fresh_speed(active_op)
|
||||
if op_speed > 0:
|
||||
return FileProgress._format_bytes(int(op_speed)) + "/s"
|
||||
|
||||
# Otherwise check other operations in priority order
|
||||
for op_key in ['download', 'extract', 'validate', 'install']:
|
||||
op_speed = _fresh_speed(op_key)
|
||||
if op_speed > 0:
|
||||
return FileProgress._format_bytes(int(op_speed)) + "/s"
|
||||
|
||||
# If engine didn't report aggregate speed, fall back to summing active file speeds
|
||||
if self.active_files:
|
||||
total_speed = sum(
|
||||
fp.speed
|
||||
for fp in self.active_files
|
||||
if fp.speed > 0 and not fp.is_complete
|
||||
)
|
||||
if total_speed > 0:
|
||||
return FileProgress._format_bytes(int(total_speed)) + "/s"
|
||||
return ""
|
||||
|
||||
def get_phase_label(self) -> str:
|
||||
"""Return a short, stable label for the current phase."""
|
||||
# Check for specific operations first (more specific than generic phase labels)
|
||||
if self.phase_name:
|
||||
phase_lower = self.phase_name.lower()
|
||||
# Check for texture conversion (very specific)
|
||||
if 'converting' in phase_lower and 'texture' in phase_lower:
|
||||
return "Converting Textures"
|
||||
# Check for BSA building
|
||||
if 'bsa' in phase_lower or ('building' in phase_lower and self.phase == InstallationPhase.INSTALL):
|
||||
return "Building BSAs"
|
||||
|
||||
# For FINALIZE phase, always prefer phase_name over generic "Finalising" label
|
||||
# This allows post-install steps to show specific labels (e.g., "Installing Wine components")
|
||||
if self.phase == InstallationPhase.FINALIZE and self.phase_name:
|
||||
return self.phase_name
|
||||
|
||||
phase_labels = {
|
||||
InstallationPhase.DOWNLOAD: "Downloading",
|
||||
InstallationPhase.EXTRACT: "Extracting",
|
||||
InstallationPhase.VALIDATE: "Validating",
|
||||
InstallationPhase.INSTALL: "Installing",
|
||||
InstallationPhase.FINALIZE: "Finalising",
|
||||
InstallationPhase.INITIALIZATION: "Preparing",
|
||||
}
|
||||
if self.phase in phase_labels:
|
||||
return phase_labels[self.phase]
|
||||
if self.phase_name:
|
||||
return self.phase_name
|
||||
if self.phase != InstallationPhase.UNKNOWN:
|
||||
return self.phase.value.title()
|
||||
return ""
|
||||
|
||||
@property
|
||||
def display_text(self) -> str:
|
||||
"""Get formatted display text for progress indicator."""
|
||||
parts = []
|
||||
|
||||
# Phase name
|
||||
phase_label = self.get_phase_label()
|
||||
if phase_label:
|
||||
parts.append(phase_label)
|
||||
|
||||
# For BSA building, show BSA count instead of generic phase progress or data progress
|
||||
if self.bsa_building_total > 0:
|
||||
# BSA building in progress - show BSA count
|
||||
parts.append(f"[{self.bsa_building_current}/{self.bsa_building_total}]")
|
||||
# Don't show data progress during BSA building (it's usually complete at 100%)
|
||||
else:
|
||||
# Normal phase - show phase progress
|
||||
phase_prog = self.phase_progress_text
|
||||
if phase_prog:
|
||||
parts.append(phase_prog)
|
||||
|
||||
# Data progress (but not during BSA building)
|
||||
data_prog = self.data_progress_text
|
||||
if data_prog:
|
||||
# Don't show if it's 100% complete (adds no value)
|
||||
if self.data_total > 0 and self.data_processed < self.data_total:
|
||||
parts.append(f"({data_prog})")
|
||||
elif self.data_total == 0 and self.data_processed > 0:
|
||||
# Show partial progress even without total
|
||||
parts.append(f"({data_prog})")
|
||||
|
||||
# Overall speed (if available, but not during BSA building)
|
||||
if self.bsa_building_total == 0:
|
||||
speed_display = self.get_overall_speed_display()
|
||||
if speed_display:
|
||||
parts.append(f"- {speed_display}")
|
||||
|
||||
# Overall percentage removed - redundant with progress bar display
|
||||
|
||||
return " ".join(parts) if parts else "Processing..."
|
||||
|
||||
def get_speed(self, operation: str) -> float:
|
||||
"""Get speed for a specific operation type."""
|
||||
return self.speeds.get(operation.lower(), 0.0)
|
||||
|
||||
def add_file(self, file_progress: FileProgress):
|
||||
"""Add or update a file in active files list."""
|
||||
# Don't re-add files that are already at 100% unless they're being actively updated
|
||||
# This prevents completed files from cluttering the list
|
||||
if file_progress.percent >= 100.0:
|
||||
# Check if this file already exists at 100%
|
||||
existing = None
|
||||
for f in self.active_files:
|
||||
if f.filename == file_progress.filename:
|
||||
existing = f
|
||||
break
|
||||
|
||||
if existing and existing.percent >= 100.0:
|
||||
# File is already at 100% - only update if it's very recent (within 0.5s)
|
||||
# This allows the completion notification to refresh the timestamp
|
||||
if time.time() - existing.last_update < 0.5:
|
||||
existing.last_update = time.time()
|
||||
# Otherwise, don't re-add it - let remove_completed_files handle cleanup
|
||||
return
|
||||
|
||||
# Remove existing entry for same filename if present
|
||||
existing = None
|
||||
for f in self.active_files:
|
||||
if f.filename == file_progress.filename:
|
||||
existing = f
|
||||
break
|
||||
|
||||
if existing:
|
||||
# Update existing entry (preserve original add time for minimum display)
|
||||
existing.operation = file_progress.operation
|
||||
existing.percent = file_progress.percent
|
||||
existing.current_size = file_progress.current_size
|
||||
existing.total_size = file_progress.total_size
|
||||
existing.speed = file_progress.speed
|
||||
existing.last_update = time.time()
|
||||
# If file just reached 100%, ensure we keep it visible for minimum time
|
||||
if file_progress.percent >= 100.0 and existing.percent < 100.0:
|
||||
# File just completed - ensure it stays visible
|
||||
existing.last_update = time.time()
|
||||
else:
|
||||
# Add new entry - set initial timestamp
|
||||
file_progress.last_update = time.time()
|
||||
self.active_files.append(file_progress)
|
||||
|
||||
# Update timestamp
|
||||
self.timestamp = time.time()
|
||||
|
||||
def remove_completed_files(self, stale_seconds: float = 0.5, stale_incomplete_seconds: float = 30.0):
|
||||
"""
|
||||
Remove files that are marked as complete, or files that haven't been updated in a while.
|
||||
|
||||
Args:
|
||||
stale_seconds: Keep completed files for this many seconds before removing (allows brief display at 100%)
|
||||
Reduced to 0.5s so tiny files that complete instantly still appear briefly
|
||||
stale_incomplete_seconds: Remove incomplete files that haven't been updated in this many seconds (handles stuck files)
|
||||
"""
|
||||
current_time = time.time()
|
||||
self.active_files = [
|
||||
f for f in self.active_files
|
||||
# Keep files that are:
|
||||
# 1. Not complete AND updated recently (active files)
|
||||
# 2. Complete AND updated very recently (show at 100% briefly so users can see all files, even tiny ones)
|
||||
if (not f.is_complete and (current_time - f.last_update) < stale_incomplete_seconds) or \
|
||||
(f.is_complete and (current_time - f.last_update) < stale_seconds)
|
||||
]
|
||||
|
||||
def update_speed(self, operation: str, speed: float):
|
||||
"""Update speed for an operation type."""
|
||||
op_key = operation.lower()
|
||||
self.speeds[op_key] = max(0.0, speed)
|
||||
self.speed_timestamps[op_key] = time.time()
|
||||
self.timestamp = time.time()
|
||||
|
||||
77
jackify/shared/steam_utils.py
Normal file
77
jackify/shared/steam_utils.py
Normal file
@@ -0,0 +1,77 @@
|
||||
"""
|
||||
Steam Utilities Module
|
||||
|
||||
Centralized Steam installation type detection to avoid redundant subprocess calls.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import subprocess
|
||||
import shutil
|
||||
from typing import Tuple
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def detect_steam_installation_types() -> Tuple[bool, bool]:
|
||||
"""
|
||||
Detect Steam installation types at startup.
|
||||
|
||||
Performs detection ONCE and returns results to be cached in SystemInfo.
|
||||
|
||||
Returns:
|
||||
Tuple[bool, bool]: (is_flatpak_steam, is_native_steam)
|
||||
"""
|
||||
is_flatpak = _detect_flatpak_steam()
|
||||
is_native = _detect_native_steam()
|
||||
|
||||
logger.info(f"Steam installation detection: Flatpak={is_flatpak}, Native={is_native}")
|
||||
|
||||
return is_flatpak, is_native
|
||||
|
||||
|
||||
def _detect_flatpak_steam() -> bool:
|
||||
"""Detect if Steam is installed as a Flatpak."""
|
||||
try:
|
||||
# First check if flatpak command exists
|
||||
if not shutil.which('flatpak'):
|
||||
return False
|
||||
|
||||
# Verify the app is actually installed (not just directory exists)
|
||||
result = subprocess.run(
|
||||
['flatpak', 'list', '--app'],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.DEVNULL, # Suppress stderr
|
||||
text=True,
|
||||
timeout=5
|
||||
)
|
||||
|
||||
if result.returncode == 0 and 'com.valvesoftware.Steam' in result.stdout:
|
||||
logger.debug("Flatpak Steam detected")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.debug(f"Error detecting Flatpak Steam: {e}")
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def _detect_native_steam() -> bool:
|
||||
"""Detect if native Steam installation exists."""
|
||||
try:
|
||||
# Check for common Steam paths
|
||||
import os
|
||||
steam_paths = [
|
||||
os.path.expanduser("~/.steam/steam"),
|
||||
os.path.expanduser("~/.local/share/Steam"),
|
||||
os.path.expanduser("~/.steam/root")
|
||||
]
|
||||
|
||||
for path in steam_paths:
|
||||
if os.path.exists(path):
|
||||
logger.debug(f"Native Steam detected at: {path}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.debug(f"Error detecting native Steam: {e}")
|
||||
|
||||
return False
|
||||
Reference in New Issue
Block a user