hyyz17200
147 lines
4.8 KiB
Python
147 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Callable, Mapping, Sequence
|
|
|
|
from .metadata import read_3mf_summary
|
|
from .models import PlateJob, ThreeMfSummary
|
|
|
|
DEFAULT_OUTPUT_PATTERN = "{plates} Plates - {sources}.3mf"
|
|
|
|
SummaryResolver = Callable[[Path], ThreeMfSummary]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OutputSummary:
|
|
plate_count: int
|
|
copy_count: int
|
|
prediction_seconds: float | None = None
|
|
weight_grams: float | None = None
|
|
filament_used_m: float | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OutputNamingOptions:
|
|
output_directory: Path | str | None = None
|
|
filename_rule: str = DEFAULT_OUTPUT_PATTERN
|
|
|
|
|
|
class SafeFormatDict(dict):
|
|
def __missing__(self, key: str) -> str:
|
|
return "{" + key + "}"
|
|
|
|
|
|
def three_mf_summary_from_mapping(data: Mapping[str, object]) -> ThreeMfSummary:
|
|
source_value = data.get("source_3mf")
|
|
return ThreeMfSummary(
|
|
source_3mf=Path(str(source_value)) if source_value is not None else Path(),
|
|
plate_count=int(data.get("plate_count") or 0),
|
|
prediction_seconds=_optional_float(data.get("prediction_seconds")),
|
|
weight_grams=_optional_float(data.get("weight_grams")),
|
|
filament_used_m=_optional_float(data.get("filament_used_m")),
|
|
filament_used_g=_optional_float(data.get("filament_used_g")),
|
|
)
|
|
|
|
|
|
def summarize_jobs_for_output(
|
|
jobs: Sequence[PlateJob],
|
|
summary_resolver: SummaryResolver = read_3mf_summary,
|
|
) -> OutputSummary:
|
|
plate_count = 0
|
|
copy_count = 0
|
|
prediction_total = 0.0
|
|
prediction_found = False
|
|
weight_total = 0.0
|
|
weight_found = False
|
|
used_m_total = 0.0
|
|
used_m_found = False
|
|
for job in jobs:
|
|
summary = summary_resolver(job.source_3mf)
|
|
copies = max(1, int(job.copies))
|
|
copy_count += copies
|
|
plate_count += summary.plate_count * copies
|
|
if summary.prediction_seconds is not None:
|
|
prediction_total += summary.prediction_seconds * copies
|
|
prediction_found = True
|
|
if summary.weight_grams is not None:
|
|
weight_total += summary.weight_grams * copies
|
|
weight_found = True
|
|
if summary.filament_used_m is not None:
|
|
used_m_total += summary.filament_used_m * copies
|
|
used_m_found = True
|
|
return OutputSummary(
|
|
plate_count=plate_count,
|
|
copy_count=copy_count,
|
|
prediction_seconds=prediction_total if prediction_found else None,
|
|
weight_grams=weight_total if weight_found else None,
|
|
filament_used_m=used_m_total if used_m_found else None,
|
|
)
|
|
|
|
|
|
def sanitize_filename(file_name: str) -> str:
|
|
sanitized = re.sub(r"[\\/:*?\"<>|]+", "_", file_name).strip()
|
|
sanitized = sanitized.rstrip(". ")
|
|
return sanitized or "packed.3mf"
|
|
|
|
|
|
def resolve_output_path(
|
|
jobs: Sequence[PlateJob],
|
|
naming: OutputNamingOptions,
|
|
summary_resolver: SummaryResolver = read_3mf_summary,
|
|
now: datetime | None = None,
|
|
) -> Path:
|
|
if not jobs:
|
|
raise ValueError("No input 3MF file was provided.")
|
|
summary = summarize_jobs_for_output(jobs, summary_resolver)
|
|
first_input = jobs[0].source_3mf
|
|
stems = [job.source_3mf.stem for job in jobs]
|
|
unique_stems = list(dict.fromkeys(stems))
|
|
source_token = first_input.stem
|
|
sources_token = source_token if len(unique_stems) == 1 else f"{source_token}_and_{len(unique_stems) - 1}_more"
|
|
timestamp = now or datetime.now()
|
|
tokens = SafeFormatDict(
|
|
source=source_token,
|
|
sources=sources_token,
|
|
plates=summary.plate_count,
|
|
copies=summary.copy_count,
|
|
date=timestamp.strftime("%Y%m%d"),
|
|
time=timestamp.strftime("%H%M%S"),
|
|
)
|
|
pattern = naming.filename_rule.strip() or DEFAULT_OUTPUT_PATTERN
|
|
file_name = sanitize_filename(pattern.format_map(tokens))
|
|
if not file_name.lower().endswith(".3mf"):
|
|
file_name += ".3mf"
|
|
output_dir_text = "" if naming.output_directory is None else str(naming.output_directory).strip()
|
|
output_dir = Path(output_dir_text) if output_dir_text else first_input.parent
|
|
return output_dir / file_name
|
|
|
|
|
|
def make_unique_for_run(path: Path, used_paths: set[Path]) -> Path:
|
|
resolved_key = path.resolve(strict=False)
|
|
if resolved_key not in used_paths:
|
|
used_paths.add(resolved_key)
|
|
return path
|
|
stem = path.stem
|
|
suffix = path.suffix
|
|
parent = path.parent
|
|
index = 2
|
|
while True:
|
|
candidate = parent / f"{stem}_{index}{suffix}"
|
|
candidate_key = candidate.resolve(strict=False)
|
|
if candidate_key not in used_paths:
|
|
used_paths.add(candidate_key)
|
|
return candidate
|
|
index += 1
|
|
|
|
|
|
def _optional_float(value: object) -> float | None:
|
|
if value is None:
|
|
return None
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|