Frank Schwenk
a60a18a253
Introduce ruff lint/format config, expand .gitignore, and reformat Python sources. Co-authored-by: Cursor <cursoragent@cursor.com>
83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
from typing import TextIO
|
|
|
|
|
|
def _format_eta(seconds: float) -> str:
|
|
seconds = max(0, int(seconds))
|
|
minutes, secs = divmod(seconds, 60)
|
|
hours, minutes = divmod(minutes, 60)
|
|
if hours:
|
|
return f"{hours}h{minutes:02d}m"
|
|
if minutes:
|
|
return f"{minutes}m{secs:02d}s"
|
|
return f"{secs}s"
|
|
|
|
|
|
class PipelineLogger:
|
|
"""Simple stdout logger for pipeline progress."""
|
|
|
|
def __init__(self, *, verbose: bool = True, stream: TextIO | None = None) -> None:
|
|
self.verbose = verbose
|
|
self.stream = stream or sys.stdout
|
|
|
|
def info(self, message: str) -> None:
|
|
if self.verbose:
|
|
print(message, file=self.stream, flush=True)
|
|
|
|
def step_start(
|
|
self,
|
|
step_index: int,
|
|
step_total: int,
|
|
step_id: str,
|
|
module_name: str,
|
|
*,
|
|
inputs: list[str],
|
|
params: dict,
|
|
) -> None:
|
|
self.info(f"Step {step_index}/{step_total}: {step_id} ({module_name})")
|
|
self.info(f" inputs: {', '.join(inputs)}")
|
|
if params:
|
|
rendered = ", ".join(f"{key}={value!r}" for key, value in params.items())
|
|
self.info(f" params: {rendered}")
|
|
|
|
def image(self, module_name: str, index: int, total: int, filename: str) -> None:
|
|
self.info(f" Applying module {module_name} to image [{index}/{total}]: {filename}")
|
|
|
|
def skipped(self, module_name: str, index: int, total: int, filename: str) -> None:
|
|
self.info(f" Skipped module {module_name} [{index}/{total}] {filename} (output exists)")
|
|
|
|
def image_done(
|
|
self,
|
|
module_name: str,
|
|
index: int,
|
|
total: int,
|
|
filename: str,
|
|
elapsed: float,
|
|
*,
|
|
eta_seconds: float | None = None,
|
|
) -> None:
|
|
eta_part = ""
|
|
if eta_seconds is not None and eta_seconds > 0:
|
|
eta_part = f" (ETA ~{_format_eta(eta_seconds)})"
|
|
self.info(
|
|
f" Finished module {module_name} [{index}/{total}] {filename} "
|
|
f"in {elapsed:.1f}s{eta_part}"
|
|
)
|
|
|
|
def warning(self, message: str) -> None:
|
|
self.info(f" Warning: {message}")
|
|
|
|
def step_done(self, step_id: str, output_dir: str, count: int) -> None:
|
|
self.info(f" Done: {count} image(s) -> {output_dir}/")
|
|
|
|
def step_skipped(self, step_id: str, count: int) -> None:
|
|
self.info(f" Skipped step {step_id} ({count} existing output(s))")
|
|
|
|
def step_reused(self, step_id: str, source_dir: str, count: int) -> None:
|
|
self.info(f" Reused external output for {step_id}: {source_dir} ({count} file(s))")
|
|
|
|
def blank(self) -> None:
|
|
self.info("")
|