475 lines
17 KiB
Python
475 lines
17 KiB
Python
from __future__ import annotations
|
|
from typing import Any
|
|
from pathlib import Path
|
|
import os
|
|
import bpy
|
|
import tempfile
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import numpy as np
|
|
|
|
from ..bpy_ns_types import NfBoolProperty, NfStringProperty
|
|
from ....registry import register_class
|
|
from .base import ExecSliceNode, SliceNode, SliceNodeJoiner, SlicingCollection, SlicingGroup
|
|
|
|
@register_class
|
|
class SLICENODE_PrinterList(ExecSliceNode):
|
|
bl_label = "Printer List"
|
|
button_label = "Start Prints"
|
|
|
|
def _refresh(self):
|
|
import json
|
|
if not self.printers_json_path:
|
|
return
|
|
|
|
abs_path = bpy.path.abspath(self.printers_json_path)
|
|
if not os.path.exists(abs_path):
|
|
return
|
|
|
|
try:
|
|
with open(abs_path, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
except Exception:
|
|
return
|
|
|
|
printers = data.get("printers", [])
|
|
printer_names = set()
|
|
|
|
for p_data in printers:
|
|
name = p_data.get("name", "Unknown")
|
|
printer_names.add(name)
|
|
sock = self.inputs.get(name)
|
|
if not sock:
|
|
self.add_input("SLICESOCK_GcodeRo", name)
|
|
|
|
# Remove stale input sockets
|
|
for sock in list(self.inputs):
|
|
if sock.name not in printer_names:
|
|
self.inputs.remove(sock)
|
|
|
|
# Remove stale output sockets (except Execution Link)
|
|
for sock in list(self.outputs):
|
|
if sock.name != "Execution Link":
|
|
self.outputs.remove(sock)
|
|
|
|
def update_config(self, context):
|
|
self._refresh()
|
|
|
|
printers_json_path: NfStringProperty(name="Config File", default="", subtype='FILE_PATH', update=update_config) # pyright: ignore[reportInvalidTypeForm]
|
|
|
|
def init(self, context):
|
|
self._init_exec()
|
|
|
|
def draw_buttons(self, context, layout):
|
|
super().draw_buttons(context, layout)
|
|
layout.prop(self, "printers_json_path")
|
|
|
|
def evaluate(self, operator=None):
|
|
import json
|
|
from ....infra.networking.printer_backends import (
|
|
PrusaLinkBackend,
|
|
CrealityBackend,
|
|
MoonrakerBackend
|
|
)
|
|
from ....infra.networking.gcode_utils import prepare_gcode_file
|
|
|
|
self._refresh()
|
|
|
|
if not self.printers_json_path:
|
|
if operator: operator.report({'ERROR'}, "No Config File path provided")
|
|
return 'error'
|
|
|
|
abs_path = bpy.path.abspath(self.printers_json_path)
|
|
if not os.path.exists(abs_path):
|
|
if operator: operator.report({'ERROR'}, f"Config file not found: {abs_path}")
|
|
return 'error'
|
|
|
|
try:
|
|
with open(abs_path, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
except Exception as e:
|
|
if operator: operator.report({'ERROR'}, f"Failed to read config file: {str(e)}")
|
|
return 'error'
|
|
|
|
printers = data.get("printers", [])
|
|
printer_configs = {p.get("name", "Unknown"): p for p in printers}
|
|
|
|
any_printed = False
|
|
for sock in self.inputs:
|
|
if sock.name not in printer_configs:
|
|
continue
|
|
|
|
gcode = self.resolve(sock.name)
|
|
if not gcode:
|
|
continue
|
|
|
|
p_data = printer_configs[sock.name]
|
|
host = str(p_data.get("ip", "localhost"))
|
|
port = int(p_data.get("port", 80))
|
|
prefix = str(p_data.get("prefix", ""))
|
|
password = str(p_data.get("password", ""))
|
|
host_type = str(p_data.get("host_type", "PRUSALINK")).upper()
|
|
|
|
# Instantiate backend
|
|
backend = None
|
|
if host_type == 'PRUSALINK':
|
|
backend = PrusaLinkBackend(host, port, prefix, api_key=password)
|
|
elif host_type == 'CREALITY':
|
|
backend = CrealityBackend(host, port, prefix)
|
|
elif host_type == 'MOONRAKER':
|
|
backend = MoonrakerBackend(host, port, prefix)
|
|
|
|
if not backend:
|
|
if operator: operator.report({'ERROR'}, f"Unsupported host type {host_type} for printer {sock.name}")
|
|
continue
|
|
|
|
# Write to temp file to send
|
|
name_part = f"NodeSlicer_{sock.name}".replace(" ", "_")
|
|
try:
|
|
file_bytes, filename = prepare_gcode_file(gcode, name_part)
|
|
except Exception as e:
|
|
if operator: operator.report({'ERROR'}, f"Failed to prepare G-code for {sock.name}: {str(e)}")
|
|
continue
|
|
|
|
suffix = Path(filename).suffix
|
|
with tempfile.NamedTemporaryFile(mode='wb', suffix=suffix, delete=False) as tf:
|
|
tf.write(file_bytes)
|
|
temp_path = Path(tf.name)
|
|
|
|
try:
|
|
backend.start_print(temp_path, filename)
|
|
if operator: operator.report({'INFO'}, f"Print started on {sock.name}")
|
|
any_printed = True
|
|
except Exception as e:
|
|
if operator: operator.report({'ERROR'}, f"Failed to start print on {sock.name}: {str(e)}")
|
|
finally:
|
|
if temp_path.exists():
|
|
temp_path.unlink()
|
|
|
|
if not any_printed:
|
|
if operator: operator.report({'WARNING'}, "No Gcode was sent to any printer")
|
|
return 'no gcode'
|
|
|
|
return 'ok'
|
|
|
|
@register_class
|
|
class SLICENODE_PrusaslicerSlice(SliceNode):
|
|
bl_label = "PrusaSlicer Slice"
|
|
|
|
prusaslicer_executable: NfStringProperty(name="PrusaSlicer Path", subtype='FILE_PATH', default='flatpak run com.prusa3d.PrusaSlicer') # pyright: ignore[reportInvalidTypeForm]
|
|
|
|
def init(self, context):
|
|
self.add_input("SLICESOCK_PrusaslicerConf", "Config")
|
|
self.add_input("SLICESOCK_Settings", "Settings")
|
|
self.add_multi_socket_group("SLICESOCK_Mesh", "SLICESOCK_GcodeRo", name="Slicing Group", in_name="Mesh", out_name="Gcode", in_display_shape='LIST')
|
|
|
|
def draw_buttons(self, context, layout):
|
|
assert layout
|
|
layout.prop(self, "prusaslicer_executable")
|
|
|
|
def evaluate_output(self, socket_name: str, identifier: str = "", **kwargs) -> Any:
|
|
if identifier.startswith("Slicing Group_ms_out_"):
|
|
try:
|
|
idx = int(identifier.split("_")[-1])
|
|
return self.evaluate_for_index(idx)
|
|
except (ValueError, IndexError):
|
|
return ""
|
|
return super().evaluate_output(socket_name, **kwargs)
|
|
|
|
def evaluate_for_index(self, index: int, operator=None) -> str:
|
|
# Optimization: Only resolve the specific mesh input for this index
|
|
target_id = f"Slicing Group_ms_in_{index}"
|
|
mesh_data: SlicingCollection | SlicingGroup = self.resolve(target_id)
|
|
|
|
conf_data = self.resolve("Config")
|
|
overrides = self.resolve("Settings")
|
|
|
|
if not conf_data:
|
|
return ""
|
|
|
|
# Extract data from config dict
|
|
printer_id = conf_data['printer']
|
|
filament_ids = conf_data['filament']
|
|
print_id = conf_data['print']
|
|
cache = conf_data['cache']
|
|
|
|
# Generate the SlicingGroup
|
|
if isinstance(mesh_data, SlicingGroup):
|
|
slicing_group = mesh_data
|
|
else:
|
|
slicing_group = SlicingGroup([])
|
|
slicing_group.collections = {mesh_data.objects[0].name: mesh_data}
|
|
|
|
# Offset by the bed center
|
|
def parse_bed_center(bed_shape_str: str) -> np.ndarray:
|
|
try:
|
|
points = []
|
|
for p in bed_shape_str.split(','):
|
|
x, y = map(float, p.split('x'))
|
|
points.append((x, y))
|
|
points_np = np.array(points)
|
|
return (points_np.min(axis=0) + points_np.max(axis=0)) / 2.0
|
|
except:
|
|
return np.array([125.0, 105.0])
|
|
|
|
bed_shape = cache.profiles[printer_id].all_conf_dict.get('bed_shape', '0x0,250x0,250x210,0x210')
|
|
bed_center = parse_bed_center(bed_shape)
|
|
|
|
a_z = -slicing_group.min_z
|
|
offset = np.array([bed_center[0], bed_center[1], a_z]) - np.array([slicing_group.center_xy[0], slicing_group.center_xy[1], 0])
|
|
slicing_group.offset(offset)
|
|
|
|
# Check for wipe_tower_position override
|
|
if overrides and "wipe_tower_position" in overrides:
|
|
wipe_tower_pos = overrides["wipe_tower_position"]
|
|
try:
|
|
wx = float(wipe_tower_pos[0])
|
|
wy = float(wipe_tower_pos[1])
|
|
slicing_group.wipe_tower_xy = np.array([wx, wy])
|
|
except Exception:
|
|
pass
|
|
|
|
# Check for offset_object override
|
|
if overrides and "offset_object" in overrides:
|
|
obj_offset = overrides["offset_object"]
|
|
try:
|
|
ox = float(obj_offset[0])
|
|
oy = float(obj_offset[1])
|
|
oz = float(obj_offset[2]) if len(obj_offset) > 2 else 0.0
|
|
slicing_group.object_offset = np.array([ox, oy, oz])
|
|
except Exception:
|
|
pass
|
|
|
|
# Prepare overrides for generate_conf_writer (excluding wipe_tower_position & offset_object)
|
|
writer_overrides = {
|
|
k: {'value': v} for k, v in (overrides or {}).items() if k not in ("wipe_tower_position", "offset_object")
|
|
}
|
|
|
|
# Generate the ConfigWriter
|
|
from ....infra.prusaslicer_profiles import ConfigWriter
|
|
conf_writer: ConfigWriter = cache.generate_conf_writer(
|
|
printer_id,
|
|
filament_ids,
|
|
print_id,
|
|
overrides=writer_overrides
|
|
)
|
|
|
|
binary_gcode_value = conf_writer.config_dict.get('binary_gcode', '0')
|
|
conf_writer.config_dict['binary_gcode'] = '0'
|
|
|
|
# Prepare 3mf path
|
|
temp_dir = Path(tempfile.gettempdir())
|
|
output_path = temp_dir / f"tmp_{index}.3mf"
|
|
|
|
try:
|
|
from ....infra._3mf_writer._3mf import prepare_3mf
|
|
prepare_3mf(output_path, slicing_group, conf_writer, z_gcodes=[])
|
|
|
|
exe_str = self.prusaslicer_executable
|
|
if not exe_str: return ""
|
|
|
|
cmd_parts = shlex.split(exe_str)
|
|
if not cmd_parts: return ""
|
|
|
|
if not shutil.which(cmd_parts[0]):
|
|
abs_path = bpy.path.abspath(cmd_parts[0])
|
|
if os.path.exists(abs_path):
|
|
cmd_parts[0] = abs_path
|
|
|
|
if not shutil.which(cmd_parts[0]):
|
|
return ""
|
|
|
|
gcode_path = output_path.with_suffix(".gcode")
|
|
cmd = cmd_parts + ["--export-gcode", str(output_path), "--output", str(gcode_path)]
|
|
|
|
process = subprocess.run(cmd, capture_output=True, text=True)
|
|
if process.returncode != 0:
|
|
return ""
|
|
|
|
if gcode_path.exists():
|
|
with open(gcode_path, 'r', encoding='utf-8') as f:
|
|
gcode_result = f.read()
|
|
gcode_result += f"\n; binary_gcode = {binary_gcode_value}\n"
|
|
|
|
output_path.unlink()
|
|
if gcode_path.exists(): gcode_path.unlink()
|
|
|
|
return gcode_result
|
|
else:
|
|
return ""
|
|
|
|
except Exception as e:
|
|
import traceback
|
|
traceback.print_exc()
|
|
return ""
|
|
|
|
@register_class
|
|
class SLICENODE_PrusaslicerPreview(ExecSliceNode):
|
|
bl_label = "PrusaSlicer Preview"
|
|
button_label = "Open Preview"
|
|
|
|
prusaslicer_executable: NfStringProperty(
|
|
name="PrusaSlicer Path",
|
|
subtype='FILE_PATH',
|
|
default='flatpak run com.prusa3d.PrusaSlicer'
|
|
) # pyright: ignore[reportInvalidTypeForm]
|
|
|
|
def init(self, context):
|
|
self._init_exec()
|
|
self.add_multi_socket_group("SLICESOCK_Gcode", name="Gcode")
|
|
|
|
def draw_buttons(self, context, layout):
|
|
super().draw_buttons(context, layout)
|
|
layout.prop(self, "prusaslicer_executable")
|
|
|
|
def evaluate(self, operator=None):
|
|
from ....infra.blender.blender_slice_nodes_generic import resolve_input
|
|
gcodes: list[str] = [g for g in self.resolve_multi_input("Gcode") if g]
|
|
if not gcodes:
|
|
if operator: operator.report({'ERROR'}, "No Gcode to preview")
|
|
return 'no gcode'
|
|
|
|
exe_str = self.prusaslicer_executable
|
|
if not exe_str:
|
|
if operator: operator.report({'ERROR'}, "PrusaSlicer executable path is empty")
|
|
return 'error'
|
|
|
|
# Parse command (handles multi-word commands like flatpak)
|
|
cmd_parts = shlex.split(exe_str)
|
|
if not cmd_parts:
|
|
if operator: operator.report({'ERROR'}, "Invalid PrusaSlicer command")
|
|
return 'error'
|
|
|
|
# Try to resolve the first part as a path if it's not a command in PATH
|
|
if not shutil.which(cmd_parts[0]):
|
|
abs_path = bpy.path.abspath(cmd_parts[0])
|
|
if os.path.exists(abs_path):
|
|
cmd_parts[0] = abs_path
|
|
|
|
# Final check if command exists
|
|
if not shutil.which(cmd_parts[0]):
|
|
if operator: operator.report({'ERROR'}, f"Command or executable not found: {cmd_parts[0]}")
|
|
return 'error'
|
|
|
|
# Write to temp files
|
|
# We use named temporary files that stay around for a bit.
|
|
# PrusaSlicer needs the files to exist when it starts.
|
|
try:
|
|
for gcode in gcodes:
|
|
fd, temp_path = tempfile.mkstemp(suffix='.gcode')
|
|
with os.fdopen(fd, 'w', encoding='utf-8') as f:
|
|
f.write(gcode)
|
|
|
|
cmd = cmd_parts + ["--gcodeviewer", temp_path]
|
|
|
|
# Use Popen to not block Blender
|
|
subprocess.Popen(cmd)
|
|
|
|
if operator:
|
|
operator.report({'INFO'}, f"PrusaSlicer Previews launched")
|
|
except Exception as e:
|
|
if operator: operator.report({'ERROR'}, f"Failed to launch preview: {str(e)}")
|
|
return 'error'
|
|
|
|
return 'ok'
|
|
|
|
@register_class
|
|
class SLICENODE_JoinExecution(SliceNodeJoiner):
|
|
bl_label = "Join Execution"
|
|
socket_type = "SLICESOCK_ExecutionLink"
|
|
|
|
def evaluate_output(self, socket_name: str, **_kwargs) -> list[ExecSliceNode]:
|
|
from ....infra.blender.blender_slice_nodes_generic import resolve_input
|
|
nodes: list[ExecSliceNode] = []
|
|
for input in self.inputs:
|
|
res = resolve_input(input)
|
|
if isinstance(res, ExecSliceNode):
|
|
nodes.append(res)
|
|
elif isinstance(res, list):
|
|
nodes.extend(res)
|
|
return nodes
|
|
|
|
@register_class
|
|
class SLICENODE_Executor(ExecSliceNode, SliceNodeJoiner):
|
|
bl_label = "Executor"
|
|
button_label = "Execute All"
|
|
socket_type = "SLICESOCK_ExecutionLink"
|
|
|
|
auto_refresh: NfBoolProperty(name="Auto-refresh", default=False) # pyright: ignore[reportInvalidTypeForm]
|
|
|
|
def init(self, context):
|
|
self._init_exec()
|
|
self.add_multi_socket_group(self.socket_type, name="Links")
|
|
|
|
def draw_buttons(self, context, layout):
|
|
super().draw_buttons(context, layout)
|
|
layout.prop(self, "auto_refresh")
|
|
|
|
def evaluate(self, operator=None):
|
|
from ....infra.blender.blender_slice_nodes_generic import (
|
|
node_execution_feedback,
|
|
feedback_disabled,
|
|
execution_cache,
|
|
resolve_input
|
|
)
|
|
|
|
# If auto_refresh is on, we don't want highlighting/redraws for upstream nodes either
|
|
maybe_disabled = feedback_disabled if self.auto_refresh else lambda: execution_cache()
|
|
|
|
for i, job in enumerate(self.inputs):
|
|
# Resolve the input to get the connected node(s)
|
|
res = resolve_input(job)
|
|
if res is None:
|
|
continue
|
|
|
|
# Ensure we have a list of jobs to process (JoinExecution returns a list)
|
|
jobs_to_run = res if isinstance(res, list) else [res]
|
|
|
|
for upstream_node in jobs_to_run:
|
|
if not hasattr(upstream_node, "evaluate"):
|
|
continue
|
|
|
|
if operator:
|
|
operator.report({'INFO'}, f"Executing: {upstream_node.bl_label}")
|
|
|
|
try:
|
|
with maybe_disabled():
|
|
with node_execution_feedback(upstream_node):
|
|
exec_res = upstream_node.evaluate(operator=operator)
|
|
if exec_res != 'ok' and exec_res is not None:
|
|
if operator: operator.report({'WARNING'}, f"Node {upstream_node.bl_label} returned {exec_res}")
|
|
except Exception as e:
|
|
if operator: operator.report({'ERROR'}, f"Failed executing {upstream_node.bl_label}: {str(e)}")
|
|
return 'error'
|
|
|
|
return 'ok'
|
|
|
|
@register_class
|
|
class SLICENODE_JoinSettings(SliceNodeJoiner):
|
|
bl_label = "Join Settings"
|
|
socket_type = "SLICESOCK_Settings"
|
|
output_socket_type = "SLICESOCK_SettingsRo"
|
|
|
|
def evaluate_output(self, socket_name: str, **_kwargs) -> dict[str, str]:
|
|
from ....infra.blender.blender_slice_nodes_generic import resolve_input
|
|
merged: dict[str, str] = {}
|
|
for input in self.inputs:
|
|
val = resolve_input(input)
|
|
if isinstance(val, dict):
|
|
merged.update(val)
|
|
return merged
|
|
|
|
@register_class
|
|
class SLICENODE_Test(ExecSliceNode):
|
|
bl_label = "Test"
|
|
|
|
button_label = "Test"
|
|
|
|
def init(self, context):
|
|
self._init_exec()
|
|
self.add_input("SLICESOCK_Gcode")
|
|
|
|
def evaluate(self, operator=None):
|
|
_ = self.resolve("Gcode")
|
|
return 'ok'
|