252 lines
8.7 KiB
Python
252 lines
8.7 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
|
|
|
|
# 默认输出文件名模板，{plates} 为总盘数，{sources} 为来源标识
|
|
DEFAULT_OUTPUT_PATTERN = "{plates}-Plates-{sources}.3mf"
|
|
|
|
# 摘要解析器类型别名，接受 Path 返回 ThreeMfSummary
|
|
SummaryResolver = Callable[[Path], ThreeMfSummary]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OutputSummary:
|
|
"""输出摘要数据类，汇总一次换板打包的整体统计信息。
|
|
|
|
属性:
|
|
plate_count: 总盘数（含份数倍增）。
|
|
copy_count: 总份数。
|
|
prediction_seconds: 预估总耗时（秒），不可用时为 None。
|
|
weight_grams: 总重量（克），不可用时为 None。
|
|
filament_used_m: 耗材总长度（米），不可用时为 None。
|
|
"""
|
|
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: 输出目录路径，None 表示使用第一个输入文件所在目录。
|
|
filename_rule: 输出文件名模板字符串。
|
|
"""
|
|
output_directory: Path | str | None = None
|
|
filename_rule: str = DEFAULT_OUTPUT_PATTERN
|
|
|
|
|
|
class SafeFormatDict(dict):
|
|
"""安全的格式化字典。
|
|
|
|
当 str.format_map 请求不存在的键时，返回原始占位符而非抛出 KeyError，
|
|
保证模板中未提供的占位符被原样保留在输出中。
|
|
"""
|
|
def __missing__(self, key: str) -> str:
|
|
return "{" + key + "}"
|
|
|
|
|
|
def three_mf_summary_from_mapping(data: Mapping[str, object]) -> ThreeMfSummary:
|
|
"""从类字典的映射对象构建 ThreeMfSummary。
|
|
|
|
用于将外部传入的配置数据（如 GUI 表格行）转换为标准摘要对象。
|
|
|
|
参数:
|
|
data: 包含摘要字段的映射对象（键名与 ThreeMfSummary 字段名对应）。
|
|
|
|
返回:
|
|
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:
|
|
"""将一组换板作业汇总为 OutputSummary。
|
|
|
|
分别计算总盘数、总份数，以及各项可选的累计统计值。
|
|
支持自定义的摘要解析器，默认为 read_3mf_summary。
|
|
|
|
参数:
|
|
jobs: PlateJob 序列。
|
|
summary_resolver: 用于获取单个 3MF 摘要的可调用对象。
|
|
|
|
返回:
|
|
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:
|
|
"""清理文件名中的非法字符，确保生成合法的文件路径。
|
|
|
|
将 Windows/Unix 中不允许出现在文件名中的字符替换为下划线，
|
|
并去掉末尾的点和空格。
|
|
|
|
参数:
|
|
file_name: 原始文件名。
|
|
|
|
返回:
|
|
str: 清理后的安全文件名，至少返回 "packed.3mf"。
|
|
"""
|
|
# 将非法字符替换为下划线
|
|
sanitized = re.sub(r"[\\/:*?\"<>|]+", "_", file_name).strip()
|
|
# 去除末尾的点号和空格（Windows 禁止）
|
|
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:
|
|
"""根据作业列表和命名选项解析最终输出文件路径。
|
|
|
|
使用模板变量（source, sources, plates, copies, date, time）替换
|
|
filename_rule 中的占位符，生成输出文件名。
|
|
|
|
参数:
|
|
jobs: PlateJob 序列。
|
|
naming: 输出命名选项。
|
|
summary_resolver: 摘要解析器，默认 read_3mf_summary。
|
|
now: 当前时间，None 时使用系统当前时间。
|
|
|
|
返回:
|
|
Path: 解析后的输出文件完整路径。
|
|
|
|
抛出:
|
|
ValueError: 若 jobs 为空。
|
|
"""
|
|
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
|
|
# 当有多个不同来源时，生成 "xxx_and_N_more" 格式的标记
|
|
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))
|
|
# 确保文件扩展名为 .3mf
|
|
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:
|
|
"""为单次运行确保输出路径唯一，通过追加递增序号避免冲突。
|
|
|
|
若目标路径已被使用，则在文件名后添加 _2、_3 等后缀直至唯一。
|
|
|
|
参数:
|
|
path: 期望的输出路径。
|
|
used_paths: 已使用路径的集合（会原地修改）。
|
|
|
|
返回:
|
|
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:
|
|
"""安全地将任意对象转换为浮点数，失败时返回 None。
|
|
|
|
参数:
|
|
value: 待转换的任意对象。
|
|
|
|
返回:
|
|
float | None: 转换成功返回浮点数，否则返回 None。
|
|
"""
|
|
if value is None:
|
|
return None
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|