Coverage for src / tracekit / utils / memory.py: 95%
268 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 23:04 +0000
1"""Memory management utilities for TraceKit.
3This module provides memory estimation, availability checking, and
4OOM prevention for large signal processing operations.
7Example:
8 >>> from tracekit.utils.memory import estimate_memory, check_memory_available
9 >>> estimate = estimate_memory('fft', samples=1e9)
10 >>> check = check_memory_available('spectrogram', samples=1e9, nperseg=4096)
12References:
13 Python psutil documentation
14"""
16from __future__ import annotations
18import gc
19import os
20import platform
21from dataclasses import dataclass
22from typing import Any
24import numpy as np
27@dataclass
28class MemoryEstimate:
29 """Memory requirement estimate for an operation.
31 Attributes:
32 data: Memory for input data (bytes).
33 intermediate: Memory for intermediate buffers (bytes).
34 output: Memory for output data (bytes).
35 total: Total memory required (bytes).
36 operation: Operation name.
37 parameters: Parameters used for estimate.
38 """
40 data: int
41 intermediate: int
42 output: int
43 total: int
44 operation: str
45 parameters: dict # type: ignore[type-arg]
47 def __repr__(self) -> str:
48 return (
49 f"MemoryEstimate({self.operation}: "
50 f"total={self.total / 1e9:.2f} GB, "
51 f"data={self.data / 1e9:.2f} GB, "
52 f"intermediate={self.intermediate / 1e9:.2f} GB, "
53 f"output={self.output / 1e9:.2f} GB)"
54 )
57@dataclass
58class MemoryCheck:
59 """Result of memory availability check.
61 Attributes:
62 sufficient: True if enough memory is available.
63 available: Available memory (bytes).
64 required: Required memory (bytes).
65 recommendation: Suggested action if insufficient.
66 """
68 sufficient: bool
69 available: int
70 required: int
71 recommendation: str
74class MemoryCheckError(Exception):
75 """Exception raised when memory check fails.
77 Attributes:
78 required: Required memory in bytes.
79 available: Available memory in bytes.
80 recommendation: Suggested action.
81 """
83 def __init__(self, message: str, required: int, available: int, recommendation: str):
84 super().__init__(message)
85 self.required = required
86 self.available = available
87 self.recommendation = recommendation
90def detect_wsl() -> bool:
91 """Detect if running in Windows Subsystem for Linux.
93 Returns:
94 True if running in WSL.
95 """
96 try:
97 with open("/proc/version") as f:
98 version = f.read().lower()
99 return "microsoft" in version or "wsl" in version
100 except (FileNotFoundError, PermissionError):
101 return False
104def get_total_memory() -> int:
105 """Get total system memory in bytes.
107 Returns:
108 Total physical memory in bytes.
109 """
110 try:
111 import psutil
113 return psutil.virtual_memory().total # type: ignore[no-any-return]
114 except ImportError:
115 # Fallback without psutil
116 if platform.system() == "Linux":
117 try:
118 with open("/proc/meminfo") as f:
119 for line in f:
120 if line.startswith("MemTotal:"):
121 # Format: "MemTotal: 16384 kB"
122 return int(line.split()[1]) * 1024
123 except (FileNotFoundError, PermissionError):
124 pass
125 # Default fallback: assume 8 GB
126 return 8 * 1024 * 1024 * 1024
129def get_available_memory() -> int:
130 """Get available memory in bytes.
132 Accounts for OS overhead and applies WSL conservative factor.
134 Returns:
135 Available memory in bytes.
136 """
137 # Get memory reserve from environment
138 reserve_str = os.environ.get("TK_MEMORY_RESERVE", "0")
139 try:
140 if reserve_str.upper().endswith("GB"):
141 reserve = int(float(reserve_str[:-2]) * 1e9)
142 elif reserve_str.upper().endswith("MB"):
143 reserve = int(float(reserve_str[:-2]) * 1e6)
144 else:
145 reserve = int(float(reserve_str))
146 except ValueError:
147 reserve = 0
149 try:
150 import psutil
152 available = psutil.virtual_memory().available
153 except ImportError:
154 # Fallback without psutil
155 if platform.system() == "Linux":
156 try:
157 with open("/proc/meminfo") as f:
158 for line in f:
159 if line.startswith("MemAvailable:"):
160 available = int(line.split()[1]) * 1024
161 break
162 else:
163 available = get_total_memory() // 2
164 except (FileNotFoundError, PermissionError):
165 available = get_total_memory() // 2
166 else:
167 available = get_total_memory() // 2
169 # Apply WSL conservative factor
170 if detect_wsl(): 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true
171 available = int(available * 0.5)
173 # Apply reserve
174 available = max(0, available - reserve)
176 return available # type: ignore[no-any-return]
179def get_swap_available() -> int:
180 """Get available swap space in bytes.
182 Returns:
183 Available swap in bytes.
184 """
185 try:
186 import psutil
188 return psutil.swap_memory().free # type: ignore[no-any-return]
189 except ImportError:
190 # Fallback
191 if platform.system() == "Linux":
192 try:
193 with open("/proc/meminfo") as f:
194 for line in f:
195 if line.startswith("SwapFree:"):
196 return int(line.split()[1]) * 1024
197 except (FileNotFoundError, PermissionError):
198 pass
199 return 0
202def get_memory_pressure() -> float:
203 """Get current memory utilization (0.0 to 1.0).
205 Returns:
206 Memory pressure as fraction of total memory used.
207 """
208 try:
209 import psutil
211 return psutil.virtual_memory().percent / 100.0 # type: ignore[no-any-return]
212 except ImportError:
213 total = get_total_memory()
214 available = get_available_memory()
215 return 1.0 - (available / total) if total > 0 else 0.5
218def estimate_memory(
219 operation: str,
220 samples: int | float | None = None,
221 *,
222 nfft: int | None = None,
223 nperseg: int | None = None,
224 noverlap: int | None = None,
225 dtype: str = "float64",
226 channels: int = 1,
227 **kwargs: Any,
228) -> MemoryEstimate:
229 """Estimate memory requirements for an operation.
231 Args:
232 operation: Operation name (fft, psd, spectrogram, eye_diagram, correlate, filter).
233 samples: Number of samples (can be float for large values).
234 nfft: FFT length (for fft, psd, spectrogram).
235 nperseg: Segment length (for spectrogram, psd).
236 noverlap: Overlap samples (for spectrogram).
237 dtype: Data type (float32 or float64).
238 channels: Number of channels.
239 **kwargs: Additional operation-specific parameters.
241 Returns:
242 MemoryEstimate with memory requirements.
244 Example:
245 >>> estimate = estimate_memory('fft', samples=1e9, nfft=8192)
246 >>> print(f"Required: {estimate.total / 1e9:.2f} GB")
247 """
248 # Bytes per element
249 bytes_per_sample = 4 if dtype == "float32" else 8
251 samples = int(samples or 0)
253 # Calculate based on operation
254 if operation == "fft":
255 nfft = nfft or _next_power_of_2(samples)
256 data_mem = samples * bytes_per_sample * channels
257 # FFT needs complex output (2x) plus work buffer
258 intermediate_mem = nfft * bytes_per_sample * 2 * 2 # complex, work buffer
259 output_mem = (nfft // 2 + 1) * bytes_per_sample * 2 * channels # complex output
261 elif operation == "psd":
262 nperseg = nperseg or 256
263 nfft = nfft or nperseg
264 data_mem = samples * bytes_per_sample * channels
265 # Welch needs segment buffer plus FFT work
266 intermediate_mem = nperseg * bytes_per_sample * 2 + nfft * bytes_per_sample * 2
267 output_mem = (nfft // 2 + 1) * bytes_per_sample * channels
269 elif operation == "spectrogram":
270 nperseg = nperseg or 256
271 noverlap = noverlap or nperseg // 2
272 nfft = nfft or nperseg
273 hop = nperseg - noverlap
274 num_segments = max(1, (samples - noverlap) // hop)
276 data_mem = samples * bytes_per_sample * channels
277 # STFT needs segment buffer
278 intermediate_mem = nperseg * bytes_per_sample * 2 + nfft * bytes_per_sample * 2
279 # Output: (nfft//2+1) frequencies x num_segments times
280 output_mem = (nfft // 2 + 1) * num_segments * bytes_per_sample * 2 * channels
282 elif operation == "eye_diagram":
283 samples_per_ui = kwargs.get("samples_per_ui", 100)
284 num_uis = kwargs.get("num_uis", 1000)
285 data_mem = samples * bytes_per_sample * channels
286 # Eye diagram accumulates traces
287 intermediate_mem = samples_per_ui * num_uis * bytes_per_sample
288 output_mem = samples_per_ui * num_uis * bytes_per_sample
290 elif operation == "correlate":
291 data_mem = samples * bytes_per_sample * 2 * channels # Two signals
292 # FFT-based correlation
293 nfft = _next_power_of_2(samples * 2)
294 intermediate_mem = nfft * bytes_per_sample * 2 * 2 # Two FFTs
295 output_mem = (samples * 2 - 1) * bytes_per_sample * channels
297 elif operation == "filter":
298 filter_order = kwargs.get("filter_order", 8)
299 data_mem = samples * bytes_per_sample * channels
300 # Filter state and buffer
301 intermediate_mem = (filter_order + samples) * bytes_per_sample
302 output_mem = samples * bytes_per_sample * channels
304 else:
305 # Generic estimate
306 data_mem = samples * bytes_per_sample * channels
307 intermediate_mem = samples * bytes_per_sample
308 output_mem = samples * bytes_per_sample * channels
310 total_mem = data_mem + intermediate_mem + output_mem
312 return MemoryEstimate(
313 data=data_mem,
314 intermediate=intermediate_mem,
315 output=output_mem,
316 total=total_mem,
317 operation=operation,
318 parameters={
319 "samples": samples,
320 "nfft": nfft,
321 "nperseg": nperseg,
322 "noverlap": noverlap,
323 "dtype": dtype,
324 "channels": channels,
325 **kwargs,
326 },
327 )
330def check_memory_available(
331 operation: str,
332 samples: int | float | None = None,
333 **kwargs: Any,
334) -> MemoryCheck:
335 """Check if sufficient memory is available for an operation.
337 Args:
338 operation: Operation name.
339 samples: Number of samples.
340 **kwargs: Additional parameters for estimate_memory.
342 Returns:
343 MemoryCheck with sufficiency status and recommendation.
345 Example:
346 >>> check = check_memory_available('spectrogram', samples=1e9, nperseg=4096)
347 >>> if not check.sufficient:
348 ... print(check.recommendation)
349 """
350 estimate = estimate_memory(operation, samples, **kwargs)
351 available = get_available_memory()
353 sufficient = estimate.total <= available
355 if sufficient:
356 recommendation = "Memory sufficient for operation."
357 else:
358 # Generate recommendations
359 ratio = estimate.total / available
360 if ratio < 2: 360 ↛ 361line 360 didn't jump to line 361 because the condition on line 360 was never true
361 recommendation = (
362 f"Need {estimate.total / 1e9:.1f} GB, have {available / 1e9:.1f} GB. "
363 "Consider closing other applications or using chunked processing."
364 )
365 elif ratio < 10:
366 recommendation = (
367 f"Need {estimate.total / 1e9:.1f} GB, have {available / 1e9:.1f} GB. "
368 f"Use chunked processing or downsample by {int(ratio)}x."
369 )
370 else:
371 recommendation = (
372 f"Need {estimate.total / 1e9:.1f} GB, have {available / 1e9:.1f} GB. "
373 "Data too large for available memory. Use streaming/chunked processing "
374 "or process a subset of the data."
375 )
377 return MemoryCheck(
378 sufficient=sufficient,
379 available=available,
380 required=estimate.total,
381 recommendation=recommendation,
382 )
385def require_memory(
386 operation: str,
387 samples: int | float | None = None,
388 **kwargs: Any,
389) -> None:
390 """Raise exception if insufficient memory for operation.
392 Args:
393 operation: Operation name.
394 samples: Number of samples.
395 **kwargs: Additional parameters.
397 Raises:
398 MemoryCheckError: If insufficient memory.
399 """
400 check = check_memory_available(operation, samples, **kwargs)
401 if not check.sufficient:
402 raise MemoryCheckError(
403 f"Insufficient memory for {operation}",
404 required=check.required,
405 available=check.available,
406 recommendation=check.recommendation,
407 )
410def _next_power_of_2(n: int) -> int:
411 """Return next power of 2 >= n."""
412 if n <= 0:
413 return 1
414 return 1 << (n - 1).bit_length()
417# Memory configuration
418_max_memory: int | None = None
421def set_max_memory(limit: int | str | None) -> None:
422 """Set global memory limit for TraceKit operations.
424 Args:
425 limit: Maximum memory in bytes, or string like "4GB", "512MB".
427 Example:
428 >>> set_max_memory("4GB")
429 >>> set_max_memory(4 * 1024 * 1024 * 1024)
430 """
431 global _max_memory
433 if limit is None:
434 _max_memory = None
435 return
437 if isinstance(limit, str):
438 limit = limit.upper().strip()
439 if limit.endswith("GB"):
440 _max_memory = int(float(limit[:-2]) * 1e9)
441 elif limit.endswith("MB"):
442 _max_memory = int(float(limit[:-2]) * 1e6)
443 elif limit.endswith("KB"): 443 ↛ 446line 443 didn't jump to line 446 because the condition on line 443 was always true
444 _max_memory = int(float(limit[:-2]) * 1e3)
445 else:
446 _max_memory = int(float(limit))
447 else:
448 _max_memory = int(limit)
451def get_max_memory() -> int:
452 """Get the current memory limit.
454 Returns:
455 Memory limit in bytes (default: 80% of available).
456 """
457 if _max_memory is not None:
458 return _max_memory
460 # Check environment variable
461 env_limit = os.environ.get("TK_MAX_MEMORY")
462 if env_limit:
463 set_max_memory(env_limit)
464 if _max_memory is not None: 464 ↛ 468line 464 didn't jump to line 468 because the condition on line 464 was always true
465 return _max_memory # type: ignore[unreachable]
467 # Default: 80% of available
468 return int(get_available_memory() * 0.8)
471def gc_collect() -> int:
472 """Force garbage collection.
474 Returns:
475 Number of unreachable objects collected.
476 """
477 return gc.collect()
480def get_memory_info() -> dict[str, int]:
481 """Get comprehensive memory information.
483 Returns:
484 Dictionary with memory statistics.
485 """
486 return {
487 "total": get_total_memory(),
488 "available": get_available_memory(),
489 "swap_available": get_swap_available(),
490 "max_memory": get_max_memory(),
491 "pressure_pct": int(get_memory_pressure() * 100),
492 "wsl": detect_wsl(),
493 }
496# ==========================================================================
497# MEM-009, MEM-010, MEM-011: Memory Configuration & Limits
498# ==========================================================================
501@dataclass
502class MemoryConfig:
503 """Global memory configuration for TraceKit operations.
506 Attributes:
507 max_memory: Global memory limit in bytes (None = 80% of available).
508 warn_threshold: Warning threshold (0.0-1.0, default 0.7).
509 critical_threshold: Critical threshold (0.0-1.0, default 0.9).
510 auto_degrade: Automatically downsample if memory exceeded.
511 """
513 max_memory: int | None = None
514 warn_threshold: float = 0.7
515 critical_threshold: float = 0.9
516 auto_degrade: bool = False
518 def __post_init__(self) -> None:
519 """Validate thresholds."""
520 if not 0.0 <= self.warn_threshold <= 1.0:
521 raise ValueError(f"warn_threshold must be 0.0-1.0, got {self.warn_threshold}")
522 if not 0.0 <= self.critical_threshold <= 1.0:
523 raise ValueError(f"critical_threshold must be 0.0-1.0, got {self.critical_threshold}")
524 if self.warn_threshold >= self.critical_threshold:
525 raise ValueError(
526 f"warn_threshold ({self.warn_threshold}) must be < critical_threshold "
527 f"({self.critical_threshold})"
528 )
531# Global memory configuration instance
532_memory_config = MemoryConfig()
535def configure_memory(
536 *,
537 max_memory: int | str | None = None,
538 warn_threshold: float | None = None,
539 critical_threshold: float | None = None,
540 auto_degrade: bool | None = None,
541) -> None:
542 """Configure global memory limits and thresholds.
545 Args:
546 max_memory: Maximum memory in bytes or string ("4GB", "512MB").
547 warn_threshold: Warning threshold (0.0-1.0).
548 critical_threshold: Critical threshold (0.0-1.0).
549 auto_degrade: Enable automatic downsampling.
551 Example:
552 >>> configure_memory(max_memory="4GB", warn_threshold=0.7, critical_threshold=0.9)
553 >>> configure_memory(auto_degrade=True)
554 """
555 global _memory_config # noqa: PLW0602
557 if max_memory is not None:
558 if isinstance(max_memory, str):
559 # Parse string format
560 limit_upper = max_memory.upper().strip()
561 if limit_upper.endswith("GB"):
562 _memory_config.max_memory = int(float(limit_upper[:-2]) * 1e9)
563 elif limit_upper.endswith("MB"): 563 ↛ 565line 563 didn't jump to line 565 because the condition on line 563 was always true
564 _memory_config.max_memory = int(float(limit_upper[:-2]) * 1e6)
565 elif limit_upper.endswith("KB"):
566 _memory_config.max_memory = int(float(limit_upper[:-2]) * 1e3)
567 else:
568 _memory_config.max_memory = int(float(limit_upper))
569 else:
570 _memory_config.max_memory = int(max_memory)
572 if warn_threshold is not None:
573 _memory_config.warn_threshold = warn_threshold
574 if critical_threshold is not None:
575 _memory_config.critical_threshold = critical_threshold
576 if auto_degrade is not None:
577 _memory_config.auto_degrade = auto_degrade
579 # Validate after updates
580 _memory_config.__post_init__()
583def get_memory_config() -> MemoryConfig:
584 """Get current memory configuration.
586 Returns:
587 Current MemoryConfig instance.
588 """
589 return _memory_config
592# ==========================================================================
593# ==========================================================================
596@dataclass
597class DownsamplingRecommendation:
598 """Recommendation for downsampling to fit memory constraints.
600 Attributes:
601 factor: Suggested downsampling factor (2, 4, 8, 16, etc.).
602 required_memory: Memory required without downsampling (bytes).
603 available_memory: Available memory (bytes).
604 new_sample_rate: Effective sample rate after downsampling (Hz).
605 message: Human-readable recommendation message.
606 """
608 factor: int
609 required_memory: int
610 available_memory: int
611 new_sample_rate: float
612 message: str
615def suggest_downsampling(
616 operation: str,
617 samples: int | float,
618 sample_rate: float,
619 **kwargs: Any,
620) -> DownsamplingRecommendation | None:
621 """Suggest downsampling factor if operation would exceed memory limits.
624 Args:
625 operation: Operation name.
626 samples: Number of samples.
627 sample_rate: Current sample rate in Hz.
628 **kwargs: Additional parameters for memory estimation.
630 Returns:
631 DownsamplingRecommendation if downsampling needed, None if sufficient memory.
633 Example:
634 >>> rec = suggest_downsampling('spectrogram', samples=1e9, sample_rate=1e9, nperseg=4096)
635 >>> if rec:
636 ... print(f"Downsample by {rec.factor}x to {rec.new_sample_rate/1e6:.1f} MSa/s")
637 """
638 estimate = estimate_memory(operation, samples, **kwargs)
639 available = get_available_memory()
641 if estimate.total <= available:
642 return None # Sufficient memory
644 # Calculate required downsampling factor
645 ratio = estimate.total / available
646 # Round up to next power of 2
647 factor = 2 ** int(np.ceil(np.log2(ratio)))
648 # Limit to reasonable factors
649 factor = min(factor, 16)
651 new_sample_rate = sample_rate / factor
652 new_samples = int(samples) // factor
654 # Re-estimate with downsampled size
655 new_estimate = estimate_memory(operation, new_samples, **kwargs)
657 message = (
658 f"Insufficient memory for {operation}. "
659 f"Need {estimate.total / 1e9:.1f} GB, have {available / 1e9:.1f} GB. "
660 f"Recommend downsampling by {factor}x (new rate: {new_sample_rate / 1e6:.1f} MSa/s). "
661 f"Estimated memory after downsampling: {new_estimate.total / 1e9:.2f} GB."
662 )
664 return DownsamplingRecommendation(
665 factor=factor,
666 required_memory=estimate.total,
667 available_memory=available,
668 new_sample_rate=new_sample_rate,
669 message=message,
670 )
673# ==========================================================================
674# ==========================================================================
677class MemoryMonitor:
678 """Context manager for monitoring memory usage and preventing OOM crashes.
681 Attributes:
682 operation: Name of the operation being monitored.
683 max_memory: Maximum allowed memory (None = use global config).
684 check_interval: How often to check memory (number of iterations).
686 Example:
687 >>> with MemoryMonitor('spectrogram', max_memory=4e9) as monitor:
688 ... for i in range(1000):
689 ... # Perform work
690 ... monitor.check(i) # Check memory periodically
691 """
693 def __init__(
694 self,
695 operation: str,
696 *,
697 max_memory: int | str | None = None,
698 check_interval: int = 100,
699 ):
700 self.operation = operation
701 self.check_interval = check_interval
702 self.start_memory = 0
703 self.peak_memory = 0
704 self.current_memory = 0
705 self._iteration = 0
707 # Parse max_memory
708 if max_memory is None:
709 self.max_memory = get_max_memory()
710 elif isinstance(max_memory, str):
711 limit_upper = max_memory.upper().strip()
712 if limit_upper.endswith("GB"):
713 self.max_memory = int(float(limit_upper[:-2]) * 1e9)
714 elif limit_upper.endswith("MB"): 714 ↛ 717line 714 didn't jump to line 717 because the condition on line 714 was always true
715 self.max_memory = int(float(limit_upper[:-2]) * 1e6)
716 else:
717 self.max_memory = int(float(limit_upper))
718 else:
719 self.max_memory = int(max_memory)
721 def __enter__(self) -> MemoryMonitor:
722 """Enter context and record starting memory."""
723 self.start_memory = self._get_process_memory()
724 self.peak_memory = self.start_memory
725 self.current_memory = self.start_memory
726 return self
728 def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
729 """Exit context."""
730 # Note: exc_val and exc_tb intentionally unused but required for Python 3.11+ compatibility
731 pass
733 def check(self, iteration: int | None = None) -> None:
734 """Check memory usage and raise error if limit approached.
736 Args:
737 iteration: Current iteration number (for periodic checking).
739 Raises:
740 MemoryError: If memory usage exceeds 95% of max_memory.
741 """
742 self._iteration += 1
744 # Only check periodically
745 if iteration is not None and iteration % self.check_interval != 0:
746 return
748 self.current_memory = self._get_process_memory()
749 self.peak_memory = max(self.peak_memory, self.current_memory)
751 # Check against available memory
752 available = get_available_memory()
753 critical_threshold = _memory_config.critical_threshold
755 if available < self.max_memory * (1 - critical_threshold): 755 ↛ 756line 755 didn't jump to line 756 because the condition on line 755 was never true
756 raise MemoryError(
757 f"Memory limit approached during {self.operation}. "
758 f"Available: {available / 1e9:.2f} GB, "
759 f"Limit: {self.max_memory / 1e9:.2f} GB. "
760 f"Operation aborted to prevent system crash."
761 )
763 def _get_process_memory(self) -> int:
764 """Get current process memory usage in bytes."""
765 try:
766 import psutil
768 process = psutil.Process()
769 return process.memory_info().rss # type: ignore[no-any-return]
770 except ImportError:
771 # Fallback: use system available memory
772 return get_total_memory() - get_available_memory()
774 def get_stats(self) -> dict[str, int]:
775 """Get memory statistics for this monitoring session.
777 Returns:
778 Dictionary with start, current, and peak memory usage.
780 Example:
781 >>> with MemoryMonitor('fft') as monitor:
782 ... # ... do work ...
783 ... stats = monitor.get_stats()
784 >>> print(f"Peak memory: {stats['peak'] / 1e6:.1f} MB")
785 """
786 return {
787 "start": self.start_memory,
788 "current": self.current_memory,
789 "peak": self.peak_memory,
790 "delta": self.peak_memory - self.start_memory,
791 }
794# ==========================================================================
795# ==========================================================================
798@dataclass
799class ProgressInfo:
800 """Progress information with memory metrics.
803 Attributes:
804 current: Current progress value.
805 total: Total progress value.
806 eta_seconds: Estimated time to completion in seconds.
807 memory_used: Current memory usage in bytes.
808 memory_peak: Peak memory usage since start in bytes.
809 operation: Name of the operation.
810 """
812 current: int
813 total: int
814 eta_seconds: float
815 memory_used: int
816 memory_peak: int
817 operation: str
819 @property
820 def percent(self) -> float:
821 """Progress percentage (0.0-100.0)."""
822 if self.total == 0:
823 return 100.0
824 return (self.current / self.total) * 100.0
826 def format_progress(self) -> str:
827 """Format progress as human-readable string.
829 Returns:
830 Formatted string like "42.5% | 1.2 GB used | 2.1 GB peak | ETA 5s"
831 """
832 return (
833 f"{self.percent:.1f}% | "
834 f"{self.memory_used / 1e9:.2f} GB used | "
835 f"{self.memory_peak / 1e9:.2f} GB peak | "
836 f"ETA {self.eta_seconds:.0f}s"
837 )