Grotti 2.0.0: eigenes Repo, Umbenennung von Farm.

grotti/parse.py

@@ -0,0 +1,455 @@
+"""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)
+_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 filaments_from_paint_info(raw):
+    try:
+        items = json.loads(raw or "")
+    except (TypeError, json.JSONDecodeError):
+        return []
+    if not isinstance(items, list):
+        return []
+    found = []
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+        try:
+            index = int(item.get("paint_index"))
+        except (TypeError, ValueError):
+            index = len(found)
+        while len(found) <= index:
+            found.append({"type": "", "color": ""})
+        found[index] = {
+            "type": str(item.get("material_type") or "").strip(),
+            "color": color_hex(item.get("paint_color")),
+        }
+    return [row for row in found if row["type"] or row["color"]]
+
+
+def filaments_of(comments):
+    painted = filaments_from_paint_info(comments.get("paint_info"))
+    if painted:
+        return painted
+    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(";")]
+    return [
+        {"type": kind, "color": colors[index] if index < len(colors) else ""}
+        for index, kind in enumerate(types)
+    ]
+
+
+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 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
+    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