diff --git a/multiace/managed/manifest.json b/multiace/managed/manifest.json index cc614e3f..c6ccd7f6 100644 --- a/multiace/managed/manifest.json +++ b/multiace/managed/manifest.json @@ -32,6 +32,7 @@ "config/extended/multiace/ace_vars.cfg", "tools/post_process_virtual_toolheads.py", "tools/job_history.py", + "tools/swap_cost.py", "i18n/", "web/backend/", "web/frontend/" diff --git a/multiace/tools/swap_cost.py b/multiace/tools/swap_cost.py index c07dc976..d90290a2 100644 --- a/multiace/tools/swap_cost.py +++ b/multiace/tools/swap_cost.py @@ -24,6 +24,8 @@ """ from __future__ import annotations +import re + # --------------------------------------------------------------------------- # Unmeasured constants # --------------------------------------------------------------------------- @@ -34,9 +36,11 @@ # that, with the shipped ace.cfg defaults (feed_speed/retract_speed 80, # load_length 2100, swap_retract_length 900, seat_overshoot_length 20, # swap_anti_ooze_retract 10), the modelled cost of an inline swap lands on -# the 210 s that post_process_virtual_toolheads.BG_SWAP_COST_INLINE_S has -# always used, and a background swap on its 30 s. Day one therefore behaves -# exactly as before and every later change is a deliberate constant edit. +# about 210 s and a background swap on 30 s - the whole-swap figures of +# post_process_virtual_toolheads.BG_SWAP_COST_INLINE_S / _BG_S. The purge +# is NOT in these seeds: build_estimate bills it on top, so an estimate with +# the default 80 mm flush reads about 28 s above the old inline figure. +# Every later change is a deliberate constant edit. # # Anything derived from a length and a speed is NOT in this table - it is # computed from the config, because that is the part we actually know. @@ -68,6 +72,350 @@ #: Kinds whose cost lands on the print's critical path in full. INLINE_KINDS = ("same_ace", "cross_ace_inline") +# Fallback densities in g/cm3, used only when the slicer wrote no +# `; filament used [g]` line (see parse_header - the Snapmaker Orca fork +# does write it, so this really is a fallback). +MATERIAL_DENSITY = { + "PLA": 1.24, "PLA+": 1.24, "PLA-CF": 1.29, "SILK": 1.24, + "PETG": 1.27, "PET": 1.27, "PETG-CF": 1.30, + "ABS": 1.04, "ASA": 1.07, "PC": 1.20, "PA": 1.14, "PA-CF": 1.20, + "TPU": 1.21, "PVA": 1.23, "HIPS": 1.04, "PP": 0.90, +} +DEFAULT_DENSITY = 1.24 +DEFAULT_DIAMETER_MM = 1.75 + +#: Purge time per swap, measured on a real machine (head mode, ACE chute): +#: about 0.166 s per mm of filament plus a fixed ~15 s of travel and wipe. +#: Not flow-limited in practice. Easy to replace once more prints are timed. +PURGE_S_PER_MM = 0.166 +PURGE_FIXED_S = 15.0 + +#: The post-processor clamps the flush-matrix stamp to this range +#: (post_process_virtual_toolheads.PURGE_MATRIX_MIN_MM / _MAX_MM). Only the +#: stamp: swap_purge_length and the stock default are used as set. +PURGE_MIN_MM = 40.0 +PURGE_MAX_MM = 150.0 + +#: A background swap primes the head with the purge length plus this much +#: (ace_bg_swap.BG_LOAD_PRIME_EXTRA), so it uses that much more filament. +BG_LOAD_PRIME_EXTRA_MM = 40.0 + +#: What the stock INNER_FLUSH_FILAMENT macro (fluidd.cfg) purges when +#: nothing tells it otherwise - `ACE_SWAP_HEAD`/`get_purge_length()` omit +#: LENGTH= and the macro's own `params.LENGTH | default(80.0)` applies. +#: This is what a head-mode inline swap costs in filament with no flush +#: stamp and swap_purge_length at its 0 default. +STOCK_FLUSH_DEFAULT_MM = 80.0 + + +_DURATION_RE = re.compile(r"(\d+(?:\.\d+)?)\s*([dhms])", re.I) + +# One line per metric, comma-separated PER EXTRUDER INDEX - not one line +# per tool. Index in the list = slicer tool index, so [g][2] is T2, and +# trailing 0.00 entries are unused extruders that must be kept as zeros or +# every tool index after them shifts. +_USED_RE = re.compile( + r"^;\s*filament used\s*\[(mm|cm3|g)\]\s*[:=]\s*(.+)$", re.I) +_TOTAL_G_RE = re.compile( + r"^;\s*total filament used\s*\[g\]\s*[:=]\s*([\d.]+)", re.I) +_LAYERS_RE = re.compile(r"^;\s*total layers count\s*[:=]\s*(\d+)", re.I) +_TIME_RE = re.compile( + r"^;\s*estimated printing time(?:\s*\([^)]*\))?\s*[:=]\s*(.+)$", re.I) +_FIRST_LAYER_TIME_RE = re.compile( + r"^;\s*estimated first layer printing time(?:\s*\([^)]*\))?\s*[:=]\s*(.+)$", + re.I) +_DENSITY_RE = re.compile(r"^;\s*filament_density\s*[:=]\s*(.+)$", re.I) +_DIAMETER_RE = re.compile(r"^;\s*filament_diameter\s*[:=]\s*(.+)$", re.I) +_TYPE_RE = re.compile(r"^;\s*filament_type\s*[:=]\s*(.+)$", re.I) + +#: Header keys that explicitly switch the prime/wipe tower on or off. +#: These are the only keys trusted to say "no tower" - an on/off switch +#: that reads false is definitive. +_TOWER_ENABLE_KEYS = ("enable_prime_tower", "prime_tower_enable", + "wipe_tower") +#: Tower GEOMETRY keys - width, brim, position. Some slicer forks +#: (Snapmaker Orca among them) write these with their last non-zero value +#: even when the tower itself is disabled, so they are consulted only when +#: the header carries no explicit on/off switch at all (see +#: purge_destination) - never used to override one that says "off". +_TOWER_GEOMETRY_KEYS = ( + "prime_tower_width", "prime_tower_brim_width", "wipe_tower_x", + "wipe_tower_width", +) +#: Header keys that mean the slicer flushed into infill/support/objects. +_FLUSH_KEYS = ( + "flush_into_infill", "flush_into_support", "flush_into_objects", +) + + +def parse_duration(text): + """'1d 2h 8m 30s' / '8m 30s' / '19s' -> seconds. None if unparseable. + + The slicer writes a DURATION STRING, not seconds, and any combination + of the four units in any order. + """ + if text is None: + return None + total = 0.0 + found = False + for value, unit in _DURATION_RE.findall(str(text)): + found = True + mult = {"d": 86400.0, "h": 3600.0, "m": 60.0, "s": 1.0}[unit.lower()] + total += float(value) * mult + return total if found else None + + +def _split_floats(text): + out = [] + for part in re.split(r"[,;]", text): + part = part.strip() + if not part: + continue + try: + out.append(float(part)) + except ValueError: + out.append(0.0) + return out + + +def _any_key_truthy(text, keys): + """True when one of `keys` appears in the header with a non-empty, + non-zero, non-false value. + + Accepts `=` OR `:` as the separator, matching what parse_color_names + in the post-processor already had to learn: slicer forks are not + consistent about it, and a separator we do not accept reads as "key + absent" - which here would mean "no prime tower" and would add a purge + the slicer had already paid for. + """ + for key in keys: + m = re.search(r"^;\s*%s\s*[:=]\s*(.+)$" % re.escape(key), text, + re.I | re.M) + if not m: + continue + for part in re.split(r"[,;]", m.group(1)): + part = part.strip().lower() + if part in ("", "0", "0.0", "false", "no", "nil", "none"): + continue + return True + return False + + +def _any_key_present(text, keys): + """True when one of `keys` appears in the header at all, whatever its + value - distinct from `_any_key_truthy`, which also asks whether that + value is "on". Used to tell "this fork wrote an explicit off-switch" + from "this fork wrote no switch at all, only geometry".""" + for key in keys: + if re.search(r"^;\s*%s\s*[:=]\s*(.+)$" % re.escape(key), text, + re.I | re.M): + return True + return False + + +def purge_destination(header_text): + """Where the SLICER'S OWN purge (if it modelled one at all) ends up: + 'tower' | 'flush' | 'mixed' | 'bin' | 'unknown'. + + Informational only - see build_estimate for why this no longer gates + whether multiACE's own ACE-swap purge gets added. A prime tower or a + flush-into-support target means the slicer printed an actual wipe + tower or flushed into actual support/infill toolpaths, consuming real + filament the post-processor never strips, so its seconds are in + `base_s` and its grams are in `; filament used [g]` ALREADY - but that + is the slicer's own mechanism, a plate feature multiACE neither needs + nor controls. The ACE swap itself still retracts to the ACE, changes + spool and purges through the hotend's own chute (INNER_FLUSH_FILAMENT) + regardless of whether the slicer also built a tower - two separate + physical events, both real, so both get reported. + + Read it from the file, never from a config default. `enable_prime_tower` + (etc.) is trusted outright, even when false - a fork can write non-zero + tower WIDTH/geometry lines whether or not the tower is enabled (seen on + Snapmaker Orca with enable_prime_tower=0), so geometry is consulted only + when the header carries no explicit on/off switch at all. When the + header carries no recognisable slicer settings at all, say 'unknown' + rather than guessing 'bin'. + """ + text = header_text or "" + if _any_key_present(text, _TOWER_ENABLE_KEYS): + tower = _any_key_truthy(text, _TOWER_ENABLE_KEYS) + else: + tower = _any_key_truthy(text, _TOWER_GEOMETRY_KEYS) + flush = _any_key_truthy(text, _FLUSH_KEYS) + if tower and flush: + return "mixed" + if tower: + return "tower" + if flush: + return "flush" + # No tower and no flush. Only trust that if the header looks like it + # carries slicer settings at all; an arbitrary gcode tells us nothing. + if re.search(r"^;\s*[a-z_]+\s*[:=]\s*", text, re.I | re.M): + return "bin" + return "unknown" + + +class HeaderInfo(object): + """Everything §1 needs out of the slicer's header block.""" + + def __init__(self): + self.per_tool_mm = [] + self.per_tool_cm3 = [] + self.per_tool_g = [] + self.total_g = None + self.layers = None + self.base_s = None + self.first_layer_s = None + self.densities = [] + self.diameters = [] + self.types = [] + self.purge_destination = "unknown" + #: 'ok' when the per-tool grams agree with the slicer's own total; + #: 'degraded' when they do not (we then trust neither silently). + self.confidence = "ok" + self.notes = [] + + def density_for(self, tool, material=None): + if tool < len(self.densities) and self.densities[tool]: + return self.densities[tool] + if material: + key = str(material).strip().upper() + if key in MATERIAL_DENSITY: + return MATERIAL_DENSITY[key] + return DEFAULT_DENSITY + + def diameter_for(self, tool): + if tool < len(self.diameters) and self.diameters[tool]: + return self.diameters[tool] + return DEFAULT_DIAMETER_MM + + def grams_for(self, tool, material=None): + """Grams printed by `tool`, from `[g]` when the slicer wrote it and + from pi*(d/2)^2*rho otherwise.""" + if tool < len(self.per_tool_g): + return self.per_tool_g[tool] + return mm_to_grams(self.mm_for(tool), self.diameter_for(tool), + self.density_for(tool, material)) + + def mm_for(self, tool): + if tool < len(self.per_tool_mm): + return self.per_tool_mm[tool] + return 0.0 + + def as_dict(self): + return { + "per_tool_mm": list(self.per_tool_mm), + "per_tool_g": list(self.per_tool_g), + "total_g": self.total_g, + "layers": self.layers, + "base_s": self.base_s, + "first_layer_s": self.first_layer_s, + "purge_destination": self.purge_destination, + "confidence": self.confidence, + "notes": list(self.notes), + } + + +def mm_to_mm3(length_mm, diameter_mm=DEFAULT_DIAMETER_MM): + r = (diameter_mm or DEFAULT_DIAMETER_MM) / 2.0 + return 3.141592653589793 * r * r * float(length_mm or 0.0) + + +def mm3_to_mm(volume_mm3, diameter_mm=DEFAULT_DIAMETER_MM): + """Inverse of mm_to_mm3 - the slicer's flush matrix is in mm3, the cost + model's purge lengths are in mm of filament.""" + r = (diameter_mm or DEFAULT_DIAMETER_MM) / 2.0 + area = 3.141592653589793 * r * r + return (float(volume_mm3 or 0.0) / area) if area else 0.0 + + +def mm_to_grams(length_mm, diameter_mm=DEFAULT_DIAMETER_MM, + density=DEFAULT_DENSITY): + return mm_to_mm3(length_mm, diameter_mm) * (density or DEFAULT_DENSITY) \ + / 1000.0 + + +def parse_header(header_text): + """Parse the slicer's metadata block into a HeaderInfo. + + Tolerant by design: a missing line means the corresponding field stays + None and the estimate degrades, never raises. + """ + info = HeaderInfo() + text = header_text or "" + for line in text.splitlines(): + s = line.strip() + if not s.startswith(";"): + continue + m = _USED_RE.match(s) + if m: + values = _split_floats(m.group(2)) + unit = m.group(1).lower() + if unit == "mm": + info.per_tool_mm = values + elif unit == "cm3": + info.per_tool_cm3 = values + else: + info.per_tool_g = values + continue + m = _TOTAL_G_RE.match(s) + if m: + info.total_g = float(m.group(1)) + continue + m = _LAYERS_RE.match(s) + if m: + info.layers = int(m.group(1)) + continue + m = _FIRST_LAYER_TIME_RE.match(s) + if m: + info.first_layer_s = parse_duration(m.group(1)) + continue + m = _TIME_RE.match(s) + if m: + info.base_s = parse_duration(m.group(1)) + continue + m = _DENSITY_RE.match(s) + if m: + info.densities = _split_floats(m.group(1)) + continue + m = _DIAMETER_RE.match(s) + if m: + info.diameters = _split_floats(m.group(1)) + continue + m = _TYPE_RE.match(s) + if m: + info.types = [p.strip() for p in re.split(r"[,;]", m.group(1)) + if p.strip()] + continue + + if not info.per_tool_g and info.per_tool_mm: + # No [g] line: fall back to pi*(d/2)^2*rho. A file without [g] + # normally has no filament_density either, so the per-material + # table is what actually carries this path. + info.per_tool_g = [ + mm_to_grams(mm, info.diameter_for(i), + info.density_for(i, info.types[i] + if i < len(info.types) else None)) + for i, mm in enumerate(info.per_tool_mm)] + info.notes.append("no `filament used [g]` - grams derived from " + "diameter and the material density table") + + # `; total filament used [g]` is an INDEPENDENT cross-check. On a + # mismatch mark the estimate degraded rather than silently trusting + # one of the two numbers. + if info.total_g is not None and info.per_tool_g: + summed = sum(info.per_tool_g) + tol = max(0.05, 0.02 * max(summed, info.total_g)) + if abs(summed - info.total_g) > tol: + info.confidence = "degraded" + info.notes.append( + "per-tool grams (%.2f) disagree with the slicer's total " + "(%.2f)" % (summed, info.total_g)) + + info.purge_destination = purge_destination(text) + return info + # --------------------------------------------------------------------------- # The cost model @@ -101,6 +449,16 @@ def __init__(self, params=None, per_ace=None, constants=None, p.get("seat_overshoot_length", 20) or 0) self.swap_anti_ooze_retract = float( p.get("swap_anti_ooze_retract", 10) or 0) + self.swap_purge_length = float(p.get("swap_purge_length", 0) or 0) + + self.swap_purge_min = float(p.get("swap_purge_min", PURGE_MIN_MM) or 0) + self.swap_purge_max = float(p.get("swap_purge_max", PURGE_MAX_MM) or 0) + #: ACE_SET_PURGE MATRIX=0 - ignore the slicer's flush-matrix stamps + #: and purge the fixed length again. + self.purge_matrix = bool(p.get("purge_matrix", True)) + self.filament_diameter = float( + p.get("filament_diameter", DEFAULT_DIAMETER_MM) + or DEFAULT_DIAMETER_MM) # -- construction --------------------------------------------------- @@ -117,7 +475,7 @@ def default(cls): "feed_speed": 80, "retract_speed": 80, "load_length": 2100, "retract_length": 1950, "swap_retract_length": 900, "seat_overshoot_length": 20, - "swap_anti_ooze_retract": 10, + "swap_anti_ooze_retract": 10, "swap_purge_length": 0, }) def with_calibration(self, swap_stats): @@ -221,3 +579,199 @@ def bg_window_minutes(self, ace=None, slot=None): + self.constants["ace_spool_change"] + self.constants["sensor_wait"]) return round((seconds * 1.25) / 60.0, 2) + + # -- purge ---------------------------------------------------------- + + def resolve_swap_purge_mm(self, stamp_mm=None): + """mm to purge for ONE swap, the way the engine does it today. + + First match wins: the flush-matrix stamp the preflight writes for + this pair (ignored when `purge_matrix` is off), else + `swap_purge_length` when above its 0 default, else the stock + 80 mm. Only the stamp is clamped (to `swap_purge_min` / + `swap_purge_max`, as the post-processor does); the other two are + used as set. Purge itself stays the slicer's business: the stamp + comes from its flush matrix. + """ + if stamp_mm and self.purge_matrix: + return self.clamp_purge_mm(stamp_mm) + if self.swap_purge_length > 0: + return self.swap_purge_length + return STOCK_FLUSH_DEFAULT_MM + + def clamp_purge_mm(self, requested_mm): + """Clamp a flush-matrix stamp to the range the post-processor uses.""" + try: + value = max(0.0, float(requested_mm or 0.0)) + except (TypeError, ValueError): + return 0.0 + if self.swap_purge_max > 0: + value = min(value, self.swap_purge_max) + return max(value, min(self.swap_purge_min, self.swap_purge_max or value)) + + def purge_seconds(self, purge_mm): + """Time for one swap's purge: per-mm extrusion plus fixed work.""" + if not purge_mm: + return 0.0 + return PURGE_FIXED_S + PURGE_S_PER_MM * float(purge_mm) + + def purge_grams(self, purge_mm, material=None): + key = str(material or "").strip().upper() + density = MATERIAL_DENSITY.get(key, DEFAULT_DENSITY) + return mm_to_grams(purge_mm, self.filament_diameter, density) + + +# --------------------------------------------------------------------------- +# The estimate block (§1.3) +# --------------------------------------------------------------------------- + +def _round(value, digits=1): + return None if value is None else round(float(value), digits) + + +def build_estimate(model, header, timeline, *, materials=None, + used_tools=None, extra_assumptions=None): + """Assemble the `estimate` block the report attaches to each plan. + + `timeline` is §3.2's per-event trace: a list of dicts with at least + `kind`, and optionally `t`, `seconds`, `purge_mm`, `purge_inline`. Everything else - + grams, per-colour breakdown, the purge total - is derived here so + backend and browser produce identical numbers. + """ + materials = materials or {} + assumptions = list(extra_assumptions or []) + + counts = {"inline": 0, "bg": 0, "unknown_window": 0, + "first_load": 0} + added_s = 0.0 + purge_mm_total = 0.0 + per_tool_purge_mm = {} + purging_swaps = 0 + added_purge_s = 0.0 + + for ev in timeline or []: + kind = ev.get("kind") + seconds = ev.get("seconds") + if seconds is None: + seconds = model.swap_seconds( + kind if kind in SWAP_KINDS else "cross_ace_inline", + ace=ev.get("ace"), slot=ev.get("slot"), + material=materials.get(ev.get("t"))) + purge = ev.get("purge_mm") + if purge is None: + # No post-processor-supplied number (an older installed copy, + # or a synthetic event built by a test) - resolve one the + # same way the rewrite would, just without a flush-matrix + # stamp to prefer (see resolve_swap_purge_mm). + purge = model.resolve_swap_purge_mm() + purge = float(purge or 0.0) + # A finished background swap purges on the parked head, off the + # critical path; its prime is the purge plus BG_LOAD_PRIME_EXTRA_MM. + # The flush is inline for every other kind, and for a background + # swap in transport mode or one that was aborted (`purge_inline`). + bg_parked = kind == "cross_ace_bg" and not ev.get("purge_inline") + if bg_parked and purge > 0: + purge += BG_LOAD_PRIME_EXTRA_MM + added_s += float(seconds or 0.0) + purge_mm_total += purge + if not bg_parked: + added_purge_s += model.purge_seconds(purge) + if purge > 0: + purging_swaps += 1 + t = ev.get("t") + if t is not None: + per_tool_purge_mm[t] = per_tool_purge_mm.get(t, 0.0) + purge + if kind == "cross_ace_bg": + counts["bg"] += 1 + elif kind == "first_load": + counts["first_load"] += 1 + else: + counts["inline"] += 1 + if ev.get("window_min") is None and kind in ("cross_ace_bg", + "cross_ace_inline"): + counts["unknown_window"] += 1 + + # The ACE-chute purge (INNER_FLUSH_FILAMENT) is a physically SEPARATE + # event from anything the slicer itself did with its own flush + # volumes - a prime tower or a flush-into-support target is real + # filament the slicer already extruded on the plate (never stripped + # by the post-processor), but the ACE swap still retracts, changes + # spool and purges through the hotend on top of that. So this is + # added to every plan's total regardless of `dest` - see + # purge_destination's docstring for why double-counting was never + # actually the risk here. + dest = header.purge_destination if header else "unknown" + if dest in ("tower", "flush", "mixed"): + label = {"tower": "prime tower", "flush": "flush into infill/support", + "mixed": "prime tower + flush"}[dest] + assumptions.append( + "%s detected - that's the slicer's own purge, separate from " + "the ACE-swap purge below" % label) + elif dest == "unknown": + assumptions.append( + "no prime-tower or flush setting found in the header - the " + "slicer's own purge destination is unknown (informational " + "only; the ACE-swap purge below is unaffected)") + if purging_swaps: + assumptions.append( + "%d ACE swap(s) purge through the chute - %.1f g added to " + "the total, on top of anything the slicer itself printed" + % (purging_swaps, sum( + model.purge_grams(mm, materials.get(t)) + for t, mm in per_tool_purge_mm.items()))) + + base_s = header.base_s if header else None + total_s = None if base_s is None else base_s + added_s + + per_color = [] + totals_mm = 0.0 + totals_g = 0.0 + tools = sorted(used_tools) if used_tools else sorted( + set(list(per_tool_purge_mm.keys()) + + list(range(len(header.per_tool_mm) if header else 0)))) + for t in tools: + material = materials.get(t) + print_mm = header.mm_for(t) if header else 0.0 + print_g = header.grams_for(t, material) if header else 0.0 + pmm = per_tool_purge_mm.get(t, 0.0) + pg = model.purge_grams(pmm, material) + # The ACE-chute purge is always extra: it is never the slicer's + # own filament-used total, whatever purge_destination says (see + # the note above). + tool_total_g = print_g + pg + totals_mm += print_mm + pmm + totals_g += tool_total_g + row = { + "t": t, + "print_mm": _round(print_mm), "print_g": _round(print_g, 2), + "purge_mm": _round(pmm), "purge_g": _round(pg, 2), + "total_m": _round((print_mm + pmm) / 1000.0, 2), + } + per_color.append(row) + purge_g_total = sum(model.purge_grams(mm, materials.get(t)) + for t, mm in per_tool_purge_mm.items()) + + confidence = model.confidence() + if header is not None and header.confidence == "degraded": + confidence = "degraded" + assumptions.extend(header.notes) + + out = { + "base_s": _round(base_s, 0), + "added_s": _round(added_s + added_purge_s, 0), + "total_s": _round(None if total_s is None + else total_s + added_purge_s, 0), + "inline_swaps": counts["inline"], + "bg_swaps": counts["bg"], + "unknown_window_swaps": counts["unknown_window"], + "purge": {"mm": _round(purge_mm_total), "g": _round(purge_g_total, 2), + "destination": dest}, + "per_color": per_color, + "totals": {"mm": _round(totals_mm), "m": _round(totals_mm / 1000.0, 2), + "g": _round(totals_g, 2)}, + "layers": header.layers if header else None, + "first_layer_s": _round(header.first_layer_s, 0) if header else None, + "confidence": confidence, + "assumptions": assumptions, + } + return out diff --git a/tests/unit/test_swap_cost.py b/tests/unit/test_swap_cost.py index 23265d12..fb55da09 100644 --- a/tests/unit/test_swap_cost.py +++ b/tests/unit/test_swap_cost.py @@ -1,4 +1,4 @@ -"""The per-swap cost model (plan E1a). +"""The cost/purge model and the slicer-header parser (plan E1). The arithmetic is derived from ace.cfg, not invented: a longer bowden really has to produce a longer estimate, and a background swap really has @@ -9,11 +9,27 @@ from multiace.tools import swap_cost as sc +# The slicer footer of a 4-colour Snapmaker Orca print, trimmed to what the +# parser reads (the shared sample gcode keeps only the first layers). +SLICER_FOOTER = """; filament used [mm] = 709.25, 298.43, 933.65, 941.90 +; filament used [cm3] = 1.71, 0.72, 2.25, 2.27 +; filament used [g] = 2.12, 0.89, 2.78, 2.81 +; total filament used [g] = 8.60 +; total layers count = 205 +; estimated printing time (normal mode) = 1h 52m 52s +; estimated first layer printing time (normal mode) = 19s +; enable_prime_tower = 1 +; filament_density = 1.24,1.24,1.24,1.24 +; filament_diameter = 1.75,1.75,1.75,1.75 +; filament_type = PLA;PLA;PLA;PLA +; flush_into_support = 1 +""" + CFG = { "feed_speed": 80, "retract_speed": 80, "load_length": 2100, "retract_length": 1950, "swap_retract_length": 900, "seat_overshoot_length": 20, - "swap_anti_ooze_retract": 10, + "swap_anti_ooze_retract": 10, "swap_purge_length": 0, } @@ -34,8 +50,9 @@ def test_terms_come_from_the_config(self, model): assert model.swap_seconds("same_ace") == pytest.approx(mech + const) def test_day_one_matches_the_old_hardcoded_costs(self, model): - """The constants were seeded so the shipped defaults reproduce - BG_SWAP_COST_INLINE_S=210 / BG_SWAP_COST_BG_S=30. If this drifts, + """The constants were seeded so the shipped defaults land on the + old whole-swap figures (BG_SWAP_COST_INLINE_S=210 / _BG_S=30), + before the purge, which the estimate bills on top. If this drifts, every existing plan silently changes shape.""" assert model.swap_seconds("cross_ace_inline") == pytest.approx(210, abs=2) @@ -115,3 +132,341 @@ def test_calibrating_one_kind_leaves_the_others_modelled(self, model): assert cal.swap_seconds("same_ace") == pytest.approx(400.0) assert cal.swap_seconds("cross_ace_bg") \ == pytest.approx(model.swap_seconds("cross_ace_bg")) + + +class TestResolveSwapPurgeMm: + """What a swap purges follows what the engine does today: the + flush-matrix stamp, else swap_purge_length, else the stock 80 mm. Only + the stamp is clamped, to 40..150 mm.""" + + def test_the_stock_default_applies_when_nothing_else_does(self, model): + assert model.resolve_swap_purge_mm() == sc.STOCK_FLUSH_DEFAULT_MM + + def test_swap_purge_length_beats_the_stock_default(self): + m = sc.SwapCostModel.from_params(dict(CFG, swap_purge_length=100)) + assert m.resolve_swap_purge_mm() == 100 + + def test_a_matrix_stamp_beats_swap_purge_length(self): + m = sc.SwapCostModel.from_params(dict(CFG, swap_purge_length=100)) + assert m.resolve_swap_purge_mm(stamp_mm=60) == 60 + + def test_matrix_off_ignores_the_stamp(self): + m = sc.SwapCostModel.from_params( + dict(CFG, swap_purge_length=100, purge_matrix=False)) + assert m.resolve_swap_purge_mm(stamp_mm=60) == 100 + + def test_only_the_stamp_is_clamped(self): + m = sc.SwapCostModel.from_params(dict(CFG, swap_purge_length=200)) + assert m.resolve_swap_purge_mm() == 200 + assert m.resolve_swap_purge_mm(stamp_mm=10) == 40 + assert m.resolve_swap_purge_mm(stamp_mm=500) == 150 + short = sc.SwapCostModel.from_params(dict(CFG, swap_purge_length=10)) + assert short.resolve_swap_purge_mm() == 10 + + def test_the_stamp_range_is_configurable(self): + m = sc.SwapCostModel.from_params( + dict(CFG, swap_purge_min=20, swap_purge_max=100)) + assert m.resolve_swap_purge_mm(stamp_mm=10) == 20 + assert m.resolve_swap_purge_mm(stamp_mm=500) == 100 + + def test_garbage_clamps_to_zero_rather_than_raising(self, model): + assert model.clamp_purge_mm("lots") == 0.0 + + def test_purge_time_is_per_mm_plus_fixed_work(self, model): + assert model.purge_seconds(100) == pytest.approx( + sc.PURGE_FIXED_S + 100 * sc.PURGE_S_PER_MM) + + def test_purge_zero_is_free(self, model): + assert model.purge_seconds(0) == 0.0 + + +class TestDurationParsing: + @pytest.mark.parametrize("text,seconds", [ + ("8m 30s", 510), + ("19s", 19), + ("5h 4m 0s", 18240), + ("1d 2h 3m 4s", 93784), + ("2h", 7200), + ]) + def test_duration_strings(self, text, seconds): + assert sc.parse_duration(text) == seconds + + def test_unparseable_is_none_not_zero(self): + """Zero would silently read as 'this print takes no time'.""" + assert sc.parse_duration("who knows") is None + assert sc.parse_duration(None) is None + + +class TestHeaderParsing: + @pytest.fixture + def header(self): + return sc.parse_header(SLICER_FOOTER) + + def test_per_extruder_lists_keep_their_index(self, header): + """One line per metric, comma-separated per extruder index - so + [g][2] is T2. Trailing zeros must be kept, or tool indices shift.""" + assert header.per_tool_mm == [709.25, 298.43, 933.65, 941.90] + assert header.per_tool_g[2] == 2.78 + + def test_trailing_unused_extruders_stay_as_zeros(self): + h = sc.parse_header("; filament used [mm] = 100.0, 0.00, 0.00, 5.00\n") + assert h.per_tool_mm == [100.0, 0.0, 0.0, 5.0] + assert h.mm_for(3) == 5.0 + + def test_printing_time_is_a_duration_not_seconds(self, header): + assert header.base_s == 6772.0 + assert header.first_layer_s == 19.0 + + def test_layer_count_is_a_free_sanity_bound(self, header): + assert header.layers == 205 + + def test_the_slicer_total_is_an_independent_cross_check(self, header): + assert header.confidence == "ok" + assert header.total_g == pytest.approx(sum(header.per_tool_g), abs=0.1) + + def test_a_mismatched_total_degrades_rather_than_picking_one(self): + h = sc.parse_header( + "; filament used [g] = 10.0, 10.0\n" + "; total filament used [g] = 90.0\n") + assert h.confidence == "degraded" + assert h.notes + + def test_total_line_is_not_mistaken_for_a_per_tool_line(self): + h = sc.parse_header( + "; filament used [g] = 1.0, 2.0\n" + "; total filament used [g] = 3.0\n") + assert h.per_tool_g == [1.0, 2.0] + assert h.total_g == 3.0 + + def test_missing_grams_falls_back_to_density(self): + h = sc.parse_header( + "; filament used [mm] = 1000.0\n" + "; filament_type = PLA\n") + expected = sc.mm_to_grams(1000.0, 1.75, sc.MATERIAL_DENSITY["PLA"]) + assert h.per_tool_g[0] == pytest.approx(expected) + assert any("density" in n for n in h.notes) + + def test_an_empty_header_never_raises(self): + h = sc.parse_header("") + assert h.base_s is None and h.per_tool_mm == [] + + def test_colon_separated_metrics_parse_too(self): + """Same fork-inconsistency lesson as purge_destination: a metric we + cannot read degrades the estimate silently, which is worse than a + visibly missing one.""" + h = sc.parse_header( + "; filament used [mm]: 100.0, 50.0\n" + "; total layers count: 42\n" + "; estimated printing time (normal mode): 1h 30m 0s\n") + assert h.per_tool_mm == [100.0, 50.0] + assert h.layers == 42 + assert h.base_s == 5400.0 + + def test_grams_for_an_unknown_material_uses_the_default_density(self): + h = sc.parse_header("; filament used [mm] = 1000.0\n") + assert h.per_tool_g[0] == pytest.approx( + sc.mm_to_grams(1000.0, 1.75, sc.DEFAULT_DENSITY)) + + +class TestPurgeDestination: + """§1.6: the one parse that changes the headline number.""" + + def test_a_prime_tower_is_detected(self): + assert sc.purge_destination( + "; enable_prime_tower = 1\n; prime_tower_width = 60\n") == "tower" + + def test_flush_into_support_is_detected(self): + assert sc.purge_destination( + "; flush_into_support = 1\n; layer_height = 0.2\n") == "flush" + + def test_both_reads_as_mixed(self): + assert sc.purge_destination( + "; enable_prime_tower = 1\n; flush_into_support = 1\n") == "mixed" + + def test_a_disabled_tower_is_not_a_tower(self): + assert sc.purge_destination( + "; enable_prime_tower = 0\n; flush_into_infill = 0\n") == "bin" + + def test_a_disabled_tower_with_leftover_geometry_is_still_not_a_tower(self): + """The actual defect this guards: Snapmaker Orca writes non-zero + `prime_tower_width`/`prime_tower_brim_width`/`wipe_tower_x` even + with `enable_prime_tower = 0` - geometry the slicer remembers for + next time, not evidence a tower was built THIS time. An explicit + enable switch, even a false one, must win over geometry.""" + assert sc.purge_destination( + "; enable_prime_tower = 0\n" + "; prime_tower_width = 30\n" + "; prime_tower_brim_width = 5\n" + "; wipe_tower_x = 23.02\n" + "; flush_into_support = 1\n") == "flush" + + def test_no_slicer_settings_at_all_is_unknown_not_bin(self): + """'cannot tell' and 'goes to the bin' are different answers.""" + assert sc.purge_destination("G1 X10 Y10\nM104 S200\n") == "unknown" + + def test_the_footer_is_a_tower_plus_flush_print(self): + assert sc.purge_destination(SLICER_FOOTER) == "mixed" + + def test_a_colon_separated_tower_is_still_a_tower(self): + """Slicer forks are not consistent about `=` vs `:` - the + post-processor's colour parser already had to learn that. A + separator we fail to accept reads as "no prime tower", and that + adds a purge the slicer already paid for.""" + assert sc.purge_destination("; enable_prime_tower: 1\n") == "tower" + + def test_a_colon_separated_flush_is_still_a_flush(self): + assert sc.purge_destination("; flush_into_support: 1\n") == "flush" + + def test_a_colon_only_header_is_not_mistaken_for_a_bin_print(self): + """'cannot tell' and 'goes to the bin' are different answers, and + this header gives no actual evidence for either.""" + assert sc.purge_destination( + "; enable_prime_tower: 1\n; layer_height: 0.2\n") != "bin" + + +class TestEstimateBlock: + def _header(self, extra=""): + return sc.parse_header( + "; filament used [mm] = 1000.0, 2000.0\n" + "; filament used [g] = 3.0, 6.0\n" + "; total filament used [g] = 9.0\n" + "; total layers count = 100\n" + "; estimated printing time (normal mode) = 1h 0m 0s\n" + + extra) + + def _timeline(self, n_inline=2, n_bg=1, purge_mm=0.0): + tl = [] + for i in range(n_inline): + tl.append({"i": i, "t": 1, "kind": "cross_ace_inline", + "window_min": 5.0, "purge_mm": purge_mm}) + for i in range(n_bg): + tl.append({"i": 100 + i, "t": 0, "kind": "cross_ace_bg", + "window_min": 9.0, "purge_mm": purge_mm}) + return tl + + def test_added_time_is_the_sum_of_the_modelled_swaps(self, model): + est = sc.build_estimate(model, self._header("; enable_prime_tower = 1\n"), + self._timeline()) + expected = (2 * model.swap_seconds("cross_ace_inline") + + model.swap_seconds("cross_ace_bg")) + assert est["added_s"] == pytest.approx(round(expected), abs=1) + assert est["total_s"] == pytest.approx(3600 + est["added_s"], abs=1) + + def test_swaps_are_counted_by_kind(self, model): + est = sc.build_estimate(model, + self._header("; enable_prime_tower = 1\n"), + self._timeline(n_inline=3, n_bg=2)) + assert est["inline_swaps"] == 3 + assert est["bg_swaps"] == 2 + + def test_a_prime_tower_purge_is_added_too(self, model): + """A prime tower is the SLICER'S OWN purge - real filament it + already extruded on the plate, never stripped by the + post-processor, and already inside base_s/[g]. The ACE swap + still retracts, changes spool and purges through the chute on + top of that: a separate physical event, so it is added + regardless of what the slicer did with its own tower.""" + header = self._header("; enable_prime_tower = 1\n") + with_purge = sc.build_estimate(model, header, + self._timeline(purge_mm=120)) + without = sc.build_estimate(model, header, self._timeline(purge_mm=0)) + assert with_purge["added_s"] > without["added_s"] + assert with_purge["purge"]["mm"] == 2 * 120 + 120 + sc.BG_LOAD_PRIME_EXTRA_MM + assert with_purge["totals"]["g"] > without["totals"]["g"] + assert any("prime tower" in a for a in with_purge["assumptions"]) + + def test_a_bin_purge_really_is_added(self, model): + header = self._header("; enable_prime_tower = 0\n") + est = sc.build_estimate(model, header, self._timeline(purge_mm=120)) + plain = sc.build_estimate(model, header, self._timeline(purge_mm=0)) + assert est["added_s"] > plain["added_s"] + + def test_an_unknown_destination_still_adds_the_ace_purge(self, model): + est = sc.build_estimate(model, sc.parse_header( + "; filament used [mm] = 1000.0\n" + "; estimated printing time = 1h 0m 0s\n"), + self._timeline(purge_mm=120)) + plain = sc.build_estimate(model, sc.parse_header( + "; filament used [mm] = 1000.0\n" + "; estimated printing time = 1h 0m 0s\n"), + self._timeline(purge_mm=0)) + assert est["purge"]["destination"] == "unknown" + assert est["added_s"] > plain["added_s"] + + def test_fewer_swaps_costs_less_time_and_filament(self, model): + """The whole point of the fix: a plan with fewer ACE swaps must + show less filament and a shorter time, not an identical + one with the purge quietly dropped from both.""" + header = self._header("; enable_prime_tower = 0\n") + many = sc.build_estimate(model, header, + self._timeline(n_inline=20, purge_mm=100), + used_tools=[0, 1]) + few = sc.build_estimate(model, header, + self._timeline(n_inline=2, purge_mm=100), + used_tools=[0, 1]) + assert many["total_s"] > few["total_s"] + assert many["totals"]["g"] > few["totals"]["g"] + + def test_purge_time_is_billed_per_swap(self, model): + header = self._header("; enable_prime_tower = 0\n") + one = sc.build_estimate(model, header, self._timeline(1, 0, 100)) + two = sc.build_estimate(model, header, self._timeline(2, 0, 100)) + per_swap = (sc.PURGE_FIXED_S + 100 * sc.PURGE_S_PER_MM + + model.swap_seconds("cross_ace_inline")) + assert one["added_s"] == pytest.approx(per_swap, abs=1) + assert two["added_s"] == pytest.approx(2 * per_swap, abs=1) + + def test_a_parked_background_purge_is_off_the_critical_path(self, model): + header = self._header("; enable_prime_tower = 0\n") + bg = [{"t": 0, "kind": "cross_ace_bg", "window_min": 9.0, + "purge_mm": 100}] + parked = sc.build_estimate(model, header, bg) + assert parked["added_s"] == pytest.approx( + model.swap_seconds("cross_ace_bg"), abs=1) + # the grams still count, with the bg prime on top of the purge + assert parked["purge"]["mm"] == 100 + sc.BG_LOAD_PRIME_EXTRA_MM + + def test_an_inline_background_purge_is_billed(self, model): + header = self._header("; enable_prime_tower = 0\n") + bg = [{"t": 0, "kind": "cross_ace_bg", "window_min": 9.0, + "purge_mm": 100, "purge_inline": True}] + est = sc.build_estimate(model, header, bg) + assert est["added_s"] == pytest.approx( + model.swap_seconds("cross_ace_bg") + model.purge_seconds(100), + abs=1) + assert est["purge"]["mm"] == 100 + + def test_per_colour_breakdown_carries_print_and_purge(self, model): + est = sc.build_estimate( + model, self._header("; enable_prime_tower = 0\n"), + self._timeline(purge_mm=100), used_tools=[0, 1]) + by_t = {row["t"]: row for row in est["per_color"]} + assert by_t[0]["print_mm"] == 1000.0 + assert by_t[1]["print_mm"] == 2000.0 + assert by_t[1]["purge_mm"] == 200 # both inline swaps target T1 + assert by_t[0]["purge_mm"] == 140 # the bg swap targets T0 (+40 prime) + + def test_the_estimate_is_never_presented_as_measured(self, model): + est = sc.build_estimate(model, self._header(), self._timeline()) + assert est["confidence"] == "modelled" + assert est["assumptions"] + + def test_a_degraded_header_degrades_the_estimate(self, model): + header = sc.parse_header( + "; filament used [g] = 1.0, 1.0\n" + "; total filament used [g] = 50.0\n" + "; estimated printing time = 1h\n") + est = sc.build_estimate(model, header, self._timeline()) + assert est["confidence"] == "degraded" + + def test_no_slicer_time_leaves_the_total_unknown_not_zero(self, model): + header = sc.parse_header("; filament used [mm] = 100.0\n") + est = sc.build_estimate(model, header, self._timeline()) + assert est["base_s"] is None + assert est["total_s"] is None + assert est["added_s"] > 0 + + def test_an_empty_timeline_adds_nothing(self, model): + est = sc.build_estimate(model, self._header(), []) + assert est["added_s"] == 0 + assert est["total_s"] == 3600