145 lines
4.3 KiB
Python
145 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import io
|
|
import logging
|
|
import shutil
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
from PIL import Image, ImageDraw
|
|
|
|
from .models import RunArtifacts
|
|
|
|
|
|
def utc_now_iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def ensure_dir(path: Path) -> None:
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def clamp(value: int, minimum: int, maximum: int) -> int:
|
|
return max(minimum, min(maximum, value))
|
|
|
|
|
|
def image_to_data_url(image: Image.Image, fmt: str = "PNG") -> str:
|
|
buf = io.BytesIO()
|
|
image.save(buf, format=fmt)
|
|
encoded = base64.b64encode(buf.getvalue()).decode("ascii")
|
|
mime = "image/png" if fmt.upper() == "PNG" else "image/jpeg"
|
|
return f"data:{mime};base64,{encoded}"
|
|
|
|
|
|
def draw_global_grid(image: Image.Image, step: int = 100) -> Image.Image:
|
|
canvas = image.convert("RGB").copy()
|
|
draw = ImageDraw.Draw(canvas)
|
|
width, height = canvas.size
|
|
|
|
grid_color = (30, 200, 255)
|
|
minor_color = (180, 220, 240)
|
|
text_bg = (0, 0, 0)
|
|
text_fg = (255, 255, 255)
|
|
|
|
draw.rectangle([0, 0, width - 1, height - 1], outline=(255, 80, 80), width=2)
|
|
|
|
for x in range(0, width, step):
|
|
color = grid_color if x % (step * 5) == 0 else minor_color
|
|
draw.line([(x, 0), (x, height)], fill=color, width=1)
|
|
label = f"x={x}"
|
|
draw.rectangle([x + 2, 2, x + 58, 18], fill=text_bg)
|
|
draw.text((x + 4, 4), label, fill=text_fg)
|
|
|
|
for y in range(0, height, step):
|
|
color = grid_color if y % (step * 5) == 0 else minor_color
|
|
draw.line([(0, y), (width, y)], fill=color, width=1)
|
|
label = f"y={y}"
|
|
draw.rectangle([2, y + 2, 58, y + 18], fill=text_bg)
|
|
draw.text((4, y + 4), label, fill=text_fg)
|
|
|
|
draw.rectangle([5, 5, 520, 35], fill=text_bg)
|
|
draw.text(
|
|
(10, 12),
|
|
"Coordinate system: origin at top-left, values in pixels",
|
|
fill=text_fg,
|
|
)
|
|
return canvas
|
|
|
|
|
|
def setup_artifacts(base_dir: Path) -> RunArtifacts:
|
|
run_id = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
root = base_dir / f"run_{run_id}"
|
|
logs_dir = root / "logs"
|
|
shots_dir = root / "screens"
|
|
enhance_dir = root / "enhanced"
|
|
for path in (root, logs_dir, shots_dir, enhance_dir):
|
|
ensure_dir(path)
|
|
return RunArtifacts(
|
|
run_id=run_id,
|
|
root_dir=root,
|
|
logs_dir=logs_dir,
|
|
shots_dir=shots_dir,
|
|
enhance_dir=enhance_dir,
|
|
log_file=logs_dir / "screenjob.log",
|
|
)
|
|
|
|
|
|
def cleanup_old_run_artifacts(base_dir: Path, retention_days: int) -> int:
|
|
if retention_days < 1 or not base_dir.exists() or not base_dir.is_dir():
|
|
return 0
|
|
cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
|
|
removed = 0
|
|
for child in base_dir.iterdir():
|
|
try:
|
|
modified_at = datetime.fromtimestamp(child.stat().st_mtime, tz=timezone.utc)
|
|
except FileNotFoundError:
|
|
continue
|
|
if modified_at >= cutoff:
|
|
continue
|
|
if child.is_dir():
|
|
shutil.rmtree(child, ignore_errors=True)
|
|
removed += 1
|
|
elif child.is_file():
|
|
try:
|
|
child.unlink()
|
|
removed += 1
|
|
except FileNotFoundError:
|
|
pass
|
|
return removed
|
|
|
|
|
|
def setup_logger(log_file: Path, verbose: bool = True) -> logging.Logger:
|
|
logger_name = f"screenjob.{log_file.parent.parent.name}"
|
|
logger = logging.getLogger(logger_name)
|
|
logger.setLevel(logging.DEBUG)
|
|
logger.propagate = False
|
|
|
|
if logger.handlers:
|
|
return logger
|
|
|
|
stream_level = logging.INFO if verbose else logging.WARNING
|
|
stream_handler = logging.StreamHandler(sys.stdout)
|
|
stream_handler.setLevel(stream_level)
|
|
stream_handler.setFormatter(
|
|
logging.Formatter(
|
|
"%(asctime)s.%(msecs)03d | %(message)s",
|
|
datefmt="%H:%M:%S",
|
|
)
|
|
)
|
|
|
|
file_handler = logging.FileHandler(log_file, encoding="utf-8")
|
|
file_handler.setLevel(logging.DEBUG)
|
|
file_handler.setFormatter(
|
|
logging.Formatter(
|
|
"%(asctime)s.%(msecs)03d | %(levelname)-8s | %(filename)s:%(lineno)d | %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
)
|
|
|
|
logger.addHandler(stream_handler)
|
|
logger.addHandler(file_handler)
|
|
return logger
|
|
|