560 lines
20 KiB
Python
560 lines
20 KiB
Python
"""Metadaten aus G-code und 3MF, ohne den Drucker anzufassen."""
|
|
|
|
import base64
|
|
import io
|
|
import json
|
|
import re
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from grotti.printers import family_from_text
|
|
|
|
THUMB_RE = re.compile(r"; thumbnail begin (\d+)x(\d+) \d+\s*(.*?); thumbnail end", re.S)
|
|
COMMENT_RE = re.compile(r"^;\s*([^:=]+?)\s*[:=]\s*(.*)$")
|
|
DURATION_RE = re.compile(r"(\d+)\s*([dhms])", re.I)
|
|
TRAILER_RE = re.compile(rb";\s*referenced_tools\s*=", re.I)
|
|
# Rinkhals mmu_ace_metadata.py: erste Zeile überspringt den vollen Scan auf dem Drucker.
|
|
RINKHALS_FINGERPRINT = b"; processed by MmuAcePatcher"
|
|
_SLICER_RE = re.compile(rb"^;.*generated by ([A-Za-z]*) .*$|^; (BambuStudio) .*$", re.I)
|
|
_TOOL_RE = re.compile(rb"^(?:MMU_CHANGE_TOOL(?:_STANDALONE)? .*?TOOL=|T)(\d{1,2})", re.I)
|
|
# Orca `; filament_colour_type` — gklib parsed wie `filament_colour` und panickt (10111).
|
|
_COLOUR_TYPE_LINE_RE = re.compile(rb"(?m)^;\s*filament_colour_type\b[^\n]*\n?")
|
|
_AUTHORIZED_SLICERS = {
|
|
"prusaslicer",
|
|
"superslicer",
|
|
"orcaslicer",
|
|
"bambustudio",
|
|
"anycubicslicernext",
|
|
}
|
|
|
|
|
|
def parse_duration(value):
|
|
total = 0
|
|
for number, unit in DURATION_RE.findall(value or ""):
|
|
total += int(number) * {"d": 86400, "h": 3600, "m": 60, "s": 1}[unit.lower()]
|
|
return total
|
|
|
|
|
|
def sum_numbers(value):
|
|
total = 0.0
|
|
for number in re.findall(r"\d+(?:[.,]\d+)?", value or ""):
|
|
total += float(number.replace(",", "."))
|
|
return total
|
|
|
|
|
|
def comments_of(text):
|
|
found = {}
|
|
for line in text.splitlines():
|
|
stripped = line.strip()
|
|
if stripped and not stripped.startswith(";"):
|
|
break
|
|
match = COMMENT_RE.match(stripped)
|
|
if match:
|
|
found[match.group(1).strip().lower()] = match.group(2).strip()
|
|
return found
|
|
|
|
|
|
def comments_scan(text):
|
|
"""Alle `; key = value`-Zeilen, auch im Footer nach G-code."""
|
|
found = {}
|
|
for line in (text or "").splitlines():
|
|
match = COMMENT_RE.match(line.strip())
|
|
if match:
|
|
found[match.group(1).strip().lower()] = match.group(2).strip()
|
|
return found
|
|
|
|
|
|
def _num_comment(comments, *keys):
|
|
for key in keys:
|
|
raw = comments.get(key) or ""
|
|
if not raw:
|
|
continue
|
|
match = re.search(r"[-+]?\d+(?:[.,]\d+)?", raw.replace(",", "."))
|
|
if match:
|
|
try:
|
|
return float(match.group(0))
|
|
except ValueError:
|
|
continue
|
|
return 0.0
|
|
|
|
|
|
def _int_comment(comments, *keys):
|
|
value = _num_comment(comments, *keys)
|
|
return int(value) if value else 0
|
|
|
|
|
|
def color_hex(channels):
|
|
if not isinstance(channels, (list, tuple)) or len(channels) < 3:
|
|
return ""
|
|
try:
|
|
rgb = [max(0, min(255, int(round(float(channel))))) for channel in channels[:3]]
|
|
except (TypeError, ValueError):
|
|
return ""
|
|
return "#%02x%02x%02x" % tuple(rgb)
|
|
|
|
|
|
def profile_name(raw):
|
|
"""Orca-Profil → Katalogname: `Sunlu PLA Meta @Kobra 3` → `Sunlu PLA Meta`."""
|
|
text = (raw or "").strip().strip('"').strip("'").strip()
|
|
if " @" in text:
|
|
text = text.rsplit(" @", 1)[0].strip()
|
|
return text
|
|
|
|
|
|
def filament_profiles(comments):
|
|
"""Profilnamen aus `; filament_settings_id`, Index = Slicer-Slot / paint_index."""
|
|
raw = (comments or {}).get("filament_settings_id") or ""
|
|
if not raw.strip():
|
|
fallback = profile_name((comments or {}).get("default_filament_profile") or "")
|
|
return [fallback] if fallback else []
|
|
return [profile_name(part) for part in raw.split(";") if profile_name(part)]
|
|
|
|
|
|
def attach_profiles(filaments, comments):
|
|
"""Hängt `profile` (Sorte) an Filament-Einträge anhand des paint_index."""
|
|
profiles = filament_profiles(comments)
|
|
if not filaments:
|
|
return filaments
|
|
if not profiles:
|
|
return filaments
|
|
for index, row in enumerate(filaments):
|
|
if not isinstance(row, dict):
|
|
continue
|
|
try:
|
|
paint = int(row.get("paint_index"))
|
|
except (TypeError, ValueError):
|
|
paint = index
|
|
name = ""
|
|
if 0 <= paint < len(profiles):
|
|
name = profiles[paint]
|
|
elif len(filaments) == 1 and len(profiles) == 1:
|
|
name = profiles[0]
|
|
if name:
|
|
row["profile"] = name
|
|
return filaments
|
|
|
|
|
|
def filaments_from_paint_info(raw):
|
|
"""Farben aus `; paint_info` — inkl. originalem `paint_index` (T-Nummer).
|
|
|
|
Bambu/Anycubic nutzen oft Lücken (T1 + T10 bei 16 Slots). Die Liste bleibt
|
|
kompakt (nur belegte Farben), aber jeder Eintrag trägt den Slicer-Index,
|
|
damit `.acm` und ACE-Map denselben Wert wie `Tn` im G-code schreiben.
|
|
"""
|
|
try:
|
|
items = json.loads(raw or "")
|
|
except (TypeError, json.JSONDecodeError):
|
|
return []
|
|
if not isinstance(items, list):
|
|
return []
|
|
found = {}
|
|
next_fallback = 0
|
|
for item in items:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
try:
|
|
index = int(item.get("paint_index"))
|
|
except (TypeError, ValueError):
|
|
index = next_fallback
|
|
if index < 0:
|
|
continue
|
|
next_fallback = max(next_fallback, index + 1)
|
|
row = {
|
|
"type": str(item.get("material_type") or "").strip(),
|
|
"color": color_hex(item.get("paint_color")),
|
|
"paint_index": index,
|
|
}
|
|
if row["type"] or row["color"]:
|
|
found[index] = row
|
|
return [found[key] for key in sorted(found)]
|
|
|
|
|
|
def filaments_of(comments):
|
|
painted = filaments_from_paint_info(comments.get("paint_info"))
|
|
if painted:
|
|
return attach_profiles(painted, comments)
|
|
types = [part.strip() for part in (comments.get("filament_type") or "").split(";") if part.strip()]
|
|
colors = [part.strip() for part in (comments.get("filament_colour") or comments.get("filament_color") or "").split(";")]
|
|
rows = [
|
|
{
|
|
"type": kind,
|
|
"color": colors[index] if index < len(colors) else "",
|
|
"paint_index": index,
|
|
}
|
|
for index, kind in enumerate(types)
|
|
]
|
|
return attach_profiles(rows, comments)
|
|
|
|
|
|
def tools_used_in_gcode(content):
|
|
"""T-/MMU-Werkzeuge aus dem G-code (wie Rinkhals mmu_ace_metadata)."""
|
|
if not content:
|
|
return set()
|
|
if isinstance(content, str):
|
|
content = content.encode("utf-8", "replace")
|
|
found = set()
|
|
for line in _gcode_lines(content):
|
|
match = _TOOL_RE.match(line)
|
|
if match:
|
|
found.add(int(match.group(1)))
|
|
return found
|
|
|
|
|
|
def _gcode_lines(content):
|
|
start = 0
|
|
size = len(content)
|
|
while start < size:
|
|
end = content.find(b"\n", start)
|
|
if end < 0:
|
|
line = content[start:]
|
|
if line.endswith(b"\r"):
|
|
line = line[:-1]
|
|
if line:
|
|
yield line
|
|
break
|
|
line = content[start:end]
|
|
if line.endswith(b"\r"):
|
|
line = line[:-1]
|
|
yield line
|
|
start = end + 1
|
|
|
|
|
|
def _paint_tool_indexes(content):
|
|
"""paint_index aus ; paint_info = […] im G-code."""
|
|
found = set()
|
|
for line in _gcode_lines(content):
|
|
if b"paint_info" not in line.lower():
|
|
continue
|
|
text = line.decode("utf-8", "replace")
|
|
match = COMMENT_RE.match(text.strip())
|
|
raw = match.group(2).strip() if match else ""
|
|
if not raw and "=" in text:
|
|
raw = text.split("=", 1)[-1].strip()
|
|
try:
|
|
items = json.loads(raw or "")
|
|
except (TypeError, json.JSONDecodeError):
|
|
continue
|
|
if not isinstance(items, list):
|
|
continue
|
|
for item in items:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
try:
|
|
found.add(int(item.get("paint_index")))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return found
|
|
|
|
|
|
def gcode_is_multicolor(content, tools=None):
|
|
"""Mehr als ein Slicer-/ACE-Tool → Multicolor (Header nicht antasten)."""
|
|
if not content or content[:2] == b"PK":
|
|
return False
|
|
indexes = set()
|
|
for value in tools or []:
|
|
try:
|
|
indexes.add(int(value))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
for line in _gcode_lines(content):
|
|
match = _TOOL_RE.match(line)
|
|
if match:
|
|
indexes.add(int(match.group(1)))
|
|
indexes |= _paint_tool_indexes(content)
|
|
return len(indexes) > 1
|
|
|
|
|
|
def strip_filament_colour_type(content, tools=None):
|
|
"""`; filament_colour_type` nur bei Einfarbig entfernen (gklib 10111).
|
|
|
|
Multicolor unverändert — ACE/paint_index und `; filament_colour` bleiben.
|
|
"""
|
|
if not content or content[:2] == b"PK":
|
|
return content
|
|
if gcode_is_multicolor(content, tools=tools):
|
|
return content
|
|
return _COLOUR_TYPE_LINE_RE.sub(b"", content)
|
|
|
|
|
|
def prepare_rinkhals_gcode(content, tools=None):
|
|
"""Kopfzeile für Rinkhals plus `; referenced_tools`, wenn sie fehlt.
|
|
|
|
`tools`: explizite Indizes (z. B. aus der ACE-Zuordnung). Sonst aus T-Befehlen
|
|
oder paint_info. Ohne Fingerprint scannt mmu_ace_metadata die ganze Datei auf
|
|
dem Drucker, bevor der Upload zurückkommt.
|
|
"""
|
|
if not content or content[:2] == b"PK":
|
|
return content
|
|
content = strip_filament_colour_type(content, tools=tools)
|
|
first_end = content.find(b"\n")
|
|
first = content if first_end < 0 else content[:first_end]
|
|
if first.endswith(b"\r"):
|
|
first = first[:-1]
|
|
has_fingerprint = first.lower().startswith(RINKHALS_FINGERPRINT.lower())
|
|
forced = set()
|
|
for value in tools or []:
|
|
try:
|
|
forced.add(int(value))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
slicer = ""
|
|
found_tools = set()
|
|
for line in _gcode_lines(content):
|
|
if not slicer and line.startswith(b";"):
|
|
match = _SLICER_RE.match(line)
|
|
if match:
|
|
raw = match.group(1) or match.group(2) or b""
|
|
slicer = raw.decode("ascii", "replace").lower()
|
|
match = _TOOL_RE.match(line)
|
|
if match:
|
|
found_tools.add(int(match.group(1)))
|
|
paint_tools = _paint_tool_indexes(content)
|
|
chosen = found_tools or forced or paint_tools
|
|
# ACE-Send mit Map, aber nichts erkannt: Einzelfarbe Platz 0.
|
|
if forced and not chosen:
|
|
chosen = {0}
|
|
stamped = content if has_fingerprint else (RINKHALS_FINGERPRINT + b"\n" + content)
|
|
need_trailer = chosen and not has_ace_trailer(content)
|
|
allow_trailer = bool(forced) or slicer in _AUTHORIZED_SLICERS
|
|
if need_trailer and allow_trailer:
|
|
if not stamped.endswith(b"\n"):
|
|
stamped += b"\n"
|
|
listed = ",".join(str(number) for number in sorted(chosen))
|
|
stamped += ("; referenced_tools = %s\n" % listed).encode("ascii")
|
|
return stamped
|
|
|
|
|
|
def has_ace_trailer(data):
|
|
"""True wenn Rinkhals-/Slicer-Trailer `; referenced_tools = …` vorhanden ist."""
|
|
if not data:
|
|
return False
|
|
if isinstance(data, str):
|
|
data = data.encode("utf-8", "replace")
|
|
if TRAILER_RE.search(data[-131072:]):
|
|
return True
|
|
return bool(TRAILER_RE.search(data[:65536]))
|
|
|
|
|
|
def content_has_ace_trailer(content, kind="", plate=""):
|
|
if has_ace_trailer(content):
|
|
return True
|
|
kind = (kind or "").lower()
|
|
if kind not in ("gcode.3mf", "3mf") and not (content and content[:2] == b"PK"):
|
|
return False
|
|
try:
|
|
with zipfile.ZipFile(io.BytesIO(content)) as archive:
|
|
names = [name for name in archive.namelist() if safe_zip_name(name)]
|
|
candidates = []
|
|
if plate and plate in names:
|
|
candidates.append(plate)
|
|
candidates.extend(
|
|
name for name in names
|
|
if name.lower().endswith(".gcode") and name not in candidates
|
|
)
|
|
for name in candidates[:3]:
|
|
with archive.open(name) as handle:
|
|
raw = handle.read()
|
|
if has_ace_trailer(raw):
|
|
return True
|
|
except (zipfile.BadZipFile, OSError, RuntimeError):
|
|
return False
|
|
return False
|
|
|
|
|
|
def _stream_has_ace_trailer(handle, head_limit=400_000):
|
|
"""Liest Stream: Kopf für Parse, Ende für Trailer — ohne ganze Datei im RAM."""
|
|
head = handle.read(head_limit)
|
|
found = bool(TRAILER_RE.search(head[:65536])) or bool(TRAILER_RE.search(head[-131072:]))
|
|
tail = head[-131072:] if len(head) > 131072 else head
|
|
while True:
|
|
chunk = handle.read(256 * 1024)
|
|
if not chunk:
|
|
break
|
|
if len(chunk) >= 131072:
|
|
tail = chunk[-131072:]
|
|
else:
|
|
tail = (tail + chunk)[-131072:]
|
|
if not found and TRAILER_RE.search(tail):
|
|
found = True
|
|
return found, head
|
|
|
|
|
|
def path_has_ace_trailer(path, kind="", plate=""):
|
|
path = Path(path)
|
|
if not path.is_file():
|
|
return False
|
|
try:
|
|
kind = (kind or "").lower()
|
|
if kind in ("gcode.3mf", "3mf") or path.name.lower().endswith(".3mf"):
|
|
if not zipfile.is_zipfile(path):
|
|
return False
|
|
with zipfile.ZipFile(path) as archive:
|
|
names = [name for name in archive.namelist() if safe_zip_name(name)]
|
|
candidates = []
|
|
if plate and plate in names:
|
|
candidates.append(plate)
|
|
candidates.extend(
|
|
name for name in names
|
|
if name.lower().endswith(".gcode") and name not in candidates
|
|
)
|
|
for name in candidates[:3]:
|
|
with archive.open(name) as handle:
|
|
found, _ = _stream_has_ace_trailer(handle)
|
|
if found:
|
|
return True
|
|
return False
|
|
size = path.stat().st_size
|
|
with path.open("rb") as handle:
|
|
if size <= 200_000:
|
|
return has_ace_trailer(handle.read())
|
|
head = handle.read(65536)
|
|
handle.seek(max(0, size - 131072))
|
|
tail = handle.read(131072)
|
|
return has_ace_trailer(head + tail)
|
|
except (OSError, zipfile.BadZipFile, RuntimeError):
|
|
return False
|
|
|
|
|
|
def parse_gcode(text, trailer=""):
|
|
head = comments_of(text)
|
|
tail = comments_scan(trailer) if trailer else comments_scan(text[-131072:] if len(text) > 200_000 else "")
|
|
comments = dict(head)
|
|
comments.update(tail)
|
|
weight_key = comments.get("total filament used [g]") or comments.get("filament used [g]") or ""
|
|
model = comments.get("printer_model") or comments.get("printer model") or ""
|
|
family = family_from_text(model) if model else family_from_text(text[:8000])
|
|
time_key = (
|
|
comments.get("estimated printing time (normal mode)")
|
|
or comments.get("estimated printing time")
|
|
or comments.get("model printing time")
|
|
or ""
|
|
)
|
|
slicer = ""
|
|
for line in (text or "").splitlines()[:40]:
|
|
match = re.search(r"generated by\s+(\S+)|;\s*(BambuStudio)\b", line, re.I)
|
|
if match:
|
|
slicer = (match.group(1) or match.group(2) or "").strip()
|
|
break
|
|
slice_meta = {
|
|
"layer_height": _num_comment(comments, "layer_height", "layer height"),
|
|
"layers": _int_comment(comments, "total layer number", "total layers count"),
|
|
"max_z_mm": _num_comment(comments, "max_z_height", "max z height", "object_height"),
|
|
"nozzle_mm": _num_comment(comments, "nozzle_diameter", "nozzle diameter"),
|
|
"temp_nozzle": _num_comment(comments, "first_layer_temperature", "nozzle_temperature"),
|
|
"temp_bed": _num_comment(comments, "bed_temperature", "first_layer_bed_temperature"),
|
|
"slicer": slicer,
|
|
}
|
|
return {
|
|
"printer_family": family,
|
|
"filaments": filaments_of(comments) or filaments_of(head),
|
|
"weight_g": sum_numbers(weight_key),
|
|
"estimated_s": parse_duration(time_key),
|
|
"thumb_entry": "embedded" if "; thumbnail begin" in text else "",
|
|
"ace_trailer": has_ace_trailer(text.encode("utf-8", "replace") if isinstance(text, str) else text),
|
|
"slice_meta": {key: value for key, value in slice_meta.items() if value},
|
|
}
|
|
|
|
|
|
def embedded_png(text):
|
|
best = None
|
|
for match in THUMB_RE.finditer(text):
|
|
width = int(match.group(1))
|
|
raw = "".join(
|
|
line[1:].strip() if line.startswith(";") else line.strip()
|
|
for line in match.group(3).splitlines()
|
|
)
|
|
try:
|
|
png = base64.b64decode(raw)
|
|
except ValueError:
|
|
continue
|
|
if png.startswith(b"\x89PNG") and (best is None or width > best[0]):
|
|
best = (width, png)
|
|
return best[1] if best else None
|
|
|
|
|
|
def safe_zip_name(name):
|
|
parts = Path(name).parts
|
|
return not name.startswith(("/", "\\")) and ".." not in parts
|
|
|
|
|
|
def pick_name(names, suffix, preferred_end):
|
|
chosen = [name for name in names if name.lower().endswith(suffix) and safe_zip_name(name)]
|
|
for name in chosen:
|
|
if name.lower().endswith(preferred_end):
|
|
return name
|
|
return sorted(chosen)[0] if chosen else ""
|
|
|
|
|
|
def _read_gcode_head_tail(path, head_limit=400_000, tail_limit=131072):
|
|
path = Path(path)
|
|
size = path.stat().st_size
|
|
with path.open("rb") as handle:
|
|
head = handle.read(head_limit).decode("utf-8", "replace")
|
|
if size <= head_limit:
|
|
return head, ""
|
|
handle.seek(max(0, size - tail_limit))
|
|
tail = handle.read(tail_limit).decode("utf-8", "replace")
|
|
return head, tail
|
|
|
|
|
|
def inspect_file(path):
|
|
path = Path(path)
|
|
empty = {
|
|
"kind": "gcode",
|
|
"printer_family": "",
|
|
"filaments": [],
|
|
"weight_g": 0,
|
|
"estimated_s": 0,
|
|
"thumb_entry": "",
|
|
"plate": "",
|
|
"ace_trailer": False,
|
|
"slice_meta": {},
|
|
}
|
|
lower = path.name.lower()
|
|
if lower.endswith(".gcode") and not lower.endswith(".3mf"):
|
|
text, trailer = _read_gcode_head_tail(path)
|
|
meta = parse_gcode(text, trailer)
|
|
meta["kind"] = "gcode"
|
|
meta["plate"] = ""
|
|
meta["ace_trailer"] = path_has_ace_trailer(path, kind="gcode")
|
|
return meta
|
|
if not zipfile.is_zipfile(path):
|
|
empty["kind"] = "3mf" if lower.endswith(".3mf") else "gcode"
|
|
return empty
|
|
with zipfile.ZipFile(path) as archive:
|
|
names = [name for name in archive.namelist() if safe_zip_name(name) and not name.startswith("__MACOSX")]
|
|
plate = pick_name(names, ".gcode", "plate_1.gcode")
|
|
thumb = pick_name(names, ".png", "plate_1.png")
|
|
text = ""
|
|
trailer = ""
|
|
ace_trailer = False
|
|
if plate:
|
|
with archive.open(plate) as handle:
|
|
ace_trailer, head = _stream_has_ace_trailer(handle)
|
|
text = head.decode("utf-8", "replace")
|
|
# Trailer: Ende der Plate-Datei nachlesen
|
|
try:
|
|
info = archive.getinfo(plate)
|
|
if info.file_size > len(head):
|
|
with archive.open(plate) as again:
|
|
again.read(max(0, info.file_size - 131072))
|
|
trailer = again.read(131072).decode("utf-8", "replace")
|
|
except (OSError, RuntimeError, KeyError):
|
|
trailer = ""
|
|
meta = parse_gcode(text, trailer) if text else dict(empty)
|
|
meta["kind"] = "gcode.3mf" if plate else "3mf"
|
|
meta["plate"] = plate
|
|
meta["thumb_entry"] = thumb or meta.get("thumb_entry") or ""
|
|
meta["ace_trailer"] = ace_trailer
|
|
if not meta["printer_family"]:
|
|
for entry in names:
|
|
if entry.lower().endswith("project_settings.config"):
|
|
raw = archive.read(entry)[:200_000]
|
|
try:
|
|
model = str(json.loads(raw).get("printer_model") or "")
|
|
except (json.JSONDecodeError, UnicodeDecodeError, AttributeError):
|
|
model = raw.decode("utf-8", "replace")
|
|
meta["printer_family"] = family_from_text(model)
|
|
break
|
|
return meta
|