352 lines
13 KiB
Python
352 lines
13 KiB
Python
from __future__ import annotations
|
|
from typing import Any
|
|
from pathlib import Path
|
|
import os
|
|
import bpy
|
|
import tempfile
|
|
import re
|
|
import json
|
|
|
|
from ..bpy_ns_types import NfStringProperty
|
|
from ....registry import register_class
|
|
from .base import ExecSliceNode, SlicePaths, SliceNode, SlicingCollection, SlicingGroup
|
|
|
|
@register_class
|
|
class SLICENODE_GcodeToFile(ExecSliceNode):
|
|
bl_label = "Gcode To File"
|
|
button_label = "Generate File"
|
|
|
|
def init(self, context):
|
|
self._init_exec()
|
|
self.add_input("SLICESOCK_DirPath", "Folder")
|
|
self.add_multi_socket_group("SLICESOCK_Gcode", name="Gcode")
|
|
|
|
def draw_buttons(self, context, layout):
|
|
super().draw_buttons(context, layout)
|
|
|
|
def evaluate(self, operator=None):
|
|
from ....infra.networking.gcode_utils import get_gcode_name_from_metadata, prepare_gcode_file
|
|
|
|
gcodes: list[str] = [gc for gc in self.resolve_multi_input('Gcode') if gc]
|
|
|
|
folder: str = self.resolve("Folder")
|
|
if not gcodes:
|
|
if operator: operator.report({'ERROR'}, "No Gcode to save")
|
|
return 'error'
|
|
if not folder:
|
|
if operator: operator.report({'ERROR'}, "No Folder provided")
|
|
return 'error'
|
|
|
|
for i, gc in enumerate(gcodes):
|
|
name = get_gcode_name_from_metadata(gc, default=f'object_{i}')
|
|
|
|
# User's requested logic: .split('.')[0]
|
|
folder_path = Path(bpy.path.abspath(f"//{folder}"))
|
|
filepath_abs = folder_path / name
|
|
dirname = os.path.dirname(filepath_abs)
|
|
basename = os.path.basename(filepath_abs)
|
|
name_part = basename.split('.')[0]
|
|
os.makedirs(dirname, exist_ok=True)
|
|
|
|
try:
|
|
file_bytes, filename = prepare_gcode_file(gc, name_part)
|
|
filepath = os.path.join(dirname, filename)
|
|
with open(filepath, "wb") as f:
|
|
f.write(file_bytes)
|
|
if filename.endswith(".bgcode"):
|
|
if operator: operator.report({'INFO'}, f"Binary G-code generated at {filepath}")
|
|
else:
|
|
if operator: operator.report({'INFO'}, f"G-code saved to {filepath}")
|
|
except Exception as e:
|
|
if operator: operator.report({'ERROR'}, str(e))
|
|
return 'error'
|
|
|
|
return 'ok'
|
|
|
|
@register_class
|
|
class SLICENODE_GcodeToTextBlock(ExecSliceNode):
|
|
bl_label = "Gcode to Text Block"
|
|
button_label = "Update Text Block"
|
|
|
|
text_block_name: NfStringProperty(name="Text Block Name", default="NodeSlicer_Gcode.gcode") # pyright: ignore[reportInvalidTypeForm]
|
|
|
|
def init(self, context):
|
|
self._init_exec()
|
|
self.add_input("SLICESOCK_Gcode")
|
|
|
|
def draw_buttons(self, context, layout):
|
|
super().draw_buttons(context, layout)
|
|
layout.prop(self, "text_block_name")
|
|
|
|
def evaluate(self, operator=None):
|
|
gcode: str = self.resolve("Gcode")
|
|
if not gcode:
|
|
return 'ok'
|
|
|
|
name = self.text_block_name
|
|
text_block = bpy.data.texts.get(name)
|
|
if text_block is None:
|
|
text_block = bpy.data.texts.new(name=name)
|
|
|
|
text_block.from_string(gcode)
|
|
|
|
# Attempt to show in an open text editor
|
|
assert bpy.context.window_manager
|
|
for window in bpy.context.window_manager.windows:
|
|
for area in window.screen.areas:
|
|
if area.type == 'TEXT_EDITOR':
|
|
for space in area.spaces:
|
|
if space.type == 'TEXT_EDITOR':
|
|
space.text = text_block
|
|
|
|
return 'ok'
|
|
|
|
@register_class
|
|
class SLICENODE_StartPrint(ExecSliceNode):
|
|
bl_label = "Start Print"
|
|
button_label = "Send to Printer"
|
|
|
|
def init(self, context):
|
|
self._init_exec()
|
|
self.add_input("SLICESOCK_Gcode")
|
|
self.add_input("SLICESOCK_Printer")
|
|
|
|
def evaluate(self, operator=None):
|
|
from ....infra.networking.printer_backends import (
|
|
PrinterConfig,
|
|
PrusaLinkBackend,
|
|
CrealityBackend,
|
|
MoonrakerBackend
|
|
)
|
|
|
|
gcode: str = self.resolve("Gcode")
|
|
config: PrinterConfig | None = self.resolve("Printer")
|
|
|
|
if not gcode:
|
|
if operator: operator.report({'ERROR'}, "No Gcode to send")
|
|
return 'no gcode'
|
|
if not config:
|
|
if operator: operator.report({'ERROR'}, "No Printer configuration")
|
|
return 'no printer'
|
|
|
|
# Instantiate correct backend
|
|
backend = None
|
|
if config.host_type == 'PRUSALINK':
|
|
backend = PrusaLinkBackend(config.host, config.port, config.prefix, api_key=config.password)
|
|
elif config.host_type == 'CREALITY':
|
|
backend = CrealityBackend(config.host, config.port, config.prefix)
|
|
elif config.host_type == 'MOONRAKER':
|
|
backend = MoonrakerBackend(config.host, config.port, config.prefix)
|
|
|
|
if not backend:
|
|
if operator: operator.report({'ERROR'}, f"Unsupported host type: {config.host_type}")
|
|
return 'unsupported'
|
|
|
|
from ....infra.networking.gcode_utils import prepare_gcode_file
|
|
|
|
# Write to temp file to send
|
|
name_part = f"NodeSlicer_{config.label}".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: {str(e)}")
|
|
return 'error'
|
|
|
|
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 {config.label}")
|
|
except Exception as e:
|
|
if operator: operator.report({'ERROR'}, f"Failed to start print: {str(e)}")
|
|
return 'error'
|
|
finally:
|
|
if temp_path.exists():
|
|
temp_path.unlink()
|
|
|
|
return 'ok'
|
|
|
|
class ExecSliceNode_DatabaseSubmit(ExecSliceNode):
|
|
record_id: NfStringProperty(name="Record ID", default="") # pyright: ignore[reportInvalidTypeForm]
|
|
|
|
_db_name = "Database"
|
|
_token_prop_name = "token_path"
|
|
|
|
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, "record_id")
|
|
|
|
def evaluate(self, operator=None):
|
|
from ....infra.networking.bgcode import convert_to_bgcode
|
|
from ....infra.networking.gcode_utils import get_gcode_name_from_metadata
|
|
|
|
gcodes: list[str] = self.resolve_multi_input("Gcode")
|
|
|
|
for gc in gcodes:
|
|
if not gc:
|
|
continue
|
|
|
|
# Parse Metadata from Gcode
|
|
fields = {}
|
|
part_name = get_gcode_name_from_metadata(gc)
|
|
fields["Part ID / Name"] = part_name
|
|
|
|
# 1. Time per print (decimal hours)
|
|
time_match = re.search(r"; estimated printing time \(normal mode\) = (.*)", gc)
|
|
if time_match:
|
|
time_str = time_match.group(1)
|
|
h = re.search(r"(\d+)h", time_str)
|
|
m = re.search(r"(\d+)m", time_str)
|
|
s = re.search(r"(\d+)s", time_str)
|
|
total_hours = 0.0
|
|
if h: total_hours += float(h.group(1))
|
|
if m: total_hours += float(m.group(1)) / 60.0
|
|
if s: total_hours += float(s.group(1)) / 3600.0
|
|
fields["Time per print"] = round(total_hours, 2)
|
|
|
|
# 2. Parts per print
|
|
parts_match = re.search(r"; objects_info = (.*)", gc)
|
|
if parts_match:
|
|
try:
|
|
info = json.loads(parts_match.group(1))
|
|
fields["Parts per print"] = len(info.get("objects", []))
|
|
except Exception: pass
|
|
|
|
# 3. Weight per print
|
|
weight_match = re.search(r"; total filament used \[g\] = ([\d\.]+)", gc)
|
|
if weight_match:
|
|
fields["Weight per print"] = int(float(weight_match.group(1)))
|
|
|
|
# 4. Profile for reference
|
|
profile_match = re.search(r"; printer_settings_id = (.*)", gc)
|
|
if profile_match:
|
|
fields["Printer Profile"] = profile_match.group(1)
|
|
|
|
# Detect binary trigger in gcode comments
|
|
is_binary = "; binary_gcode = 1" in gc
|
|
|
|
filename = part_name + (".bgcode" if is_binary else ".gcode")
|
|
|
|
token_path = getattr(self, self._token_prop_name, "")
|
|
token = ""
|
|
if token_path:
|
|
abs_token_path = bpy.path.abspath(token_path)
|
|
if os.path.exists(abs_token_path):
|
|
try:
|
|
with open(abs_token_path, 'r', encoding='utf-8') as f:
|
|
token = f.read().strip()
|
|
except Exception as e:
|
|
if operator: operator.report({'ERROR'}, f"Failed to read token file: {str(e)}")
|
|
return 'error'
|
|
|
|
if not self._check_config(token):
|
|
if operator: operator.report({'ERROR'}, f"{self._db_name} configuration incomplete")
|
|
return 'incomplete config'
|
|
|
|
temp_dir = tempfile.TemporaryDirectory()
|
|
temp_path = Path(temp_dir.name) / filename
|
|
|
|
try:
|
|
if is_binary:
|
|
content = convert_to_bgcode(gc, part_name, True)
|
|
with open(temp_path, "wb") as f:
|
|
f.write(content)
|
|
else:
|
|
with open(temp_path, 'w', encoding='utf-8') as f:
|
|
f.write(gc)
|
|
|
|
record_id = self._submit_to_db(token, fields, temp_path)
|
|
if operator: operator.report({'INFO'}, f"Submitted to {self._db_name} (Record: {record_id})")
|
|
except Exception as e:
|
|
if operator: operator.report({'ERROR'}, f"{self._db_name} Error: {str(e)}")
|
|
finally:
|
|
if temp_path.exists():
|
|
temp_path.unlink()
|
|
temp_dir.cleanup()
|
|
|
|
def _check_config(self, token: str) -> bool:
|
|
return False
|
|
|
|
def _submit_to_db(self, token: str, fields: dict, temp_path: Path) -> str:
|
|
return ""
|
|
|
|
@register_class
|
|
class SLICENODE_BaserowSubmit(ExecSliceNode_DatabaseSubmit):
|
|
bl_label = "Baserow Submit"
|
|
button_label = "Submit to Baserow"
|
|
|
|
host: NfStringProperty(name="Host", default="https://baserow.io") # pyright: ignore[reportInvalidTypeForm]
|
|
token_path: NfStringProperty(name="Token File", subtype='FILE_PATH', default="") # pyright: ignore[reportInvalidTypeForm]
|
|
table_id: NfStringProperty(name="Table ID", default="") # pyright: ignore[reportInvalidTypeForm]
|
|
|
|
_db_name = "Baserow"
|
|
_token_prop_name = "token_path"
|
|
|
|
def draw_buttons(self, context, layout):
|
|
layout.prop(self, "host")
|
|
layout.prop(self, "token_path")
|
|
layout.prop(self, "table_id")
|
|
super().draw_buttons(context, layout)
|
|
|
|
def _check_config(self, token):
|
|
return bool(token and self.table_id and self.host)
|
|
|
|
def _submit_to_db(self, token, fields, temp_path):
|
|
from ....infra.networking.baserow_backend import BaserowBackend
|
|
db = BaserowBackend(self.host, token)
|
|
|
|
if self.record_id:
|
|
try:
|
|
rid = int(self.record_id)
|
|
res = db.upsert_record(self.table_id, rid, fields)
|
|
except ValueError:
|
|
raise ValueError("Record ID must be an integer")
|
|
else:
|
|
res = db.create_record(self.table_id, fields)
|
|
|
|
record_id = int(res['id'])
|
|
db.upload_attachment(self.table_id, record_id, 'GCODE', temp_path)
|
|
return str(record_id)
|
|
|
|
@register_class
|
|
class SLICENODE_AirtableSubmit(ExecSliceNode_DatabaseSubmit):
|
|
bl_label = "Airtable Submit"
|
|
button_label = "Submit to Airtable"
|
|
|
|
pat_path: NfStringProperty(name="PAT File", subtype='FILE_PATH', default="") # pyright: ignore[reportInvalidTypeForm]
|
|
base_id: NfStringProperty(name="Base ID", default="") # pyright: ignore[reportInvalidTypeForm]
|
|
table_id: NfStringProperty(name="Table ID", default="") # pyright: ignore[reportInvalidTypeForm]
|
|
|
|
_db_name = "Airtable"
|
|
_token_prop_name = "pat_path"
|
|
|
|
def draw_buttons(self, context, layout):
|
|
layout.prop(self, "pat_path")
|
|
layout.prop(self, "base_id")
|
|
layout.prop(self, "table_id")
|
|
super().draw_buttons(context, layout)
|
|
|
|
def _check_config(self, token):
|
|
return bool(token and self.base_id and self.table_id)
|
|
|
|
def _submit_to_db(self, token, fields, temp_path):
|
|
from ....infra.networking.airtable_backend import AirtableInterface, AirtableRecord
|
|
db = AirtableInterface(token)
|
|
|
|
if self.record_id:
|
|
record: AirtableRecord = {"id": self.record_id, "fields": fields}
|
|
res = db.update_records(self.base_id, self.table_id, [record])
|
|
record_id = res[0]['id']
|
|
else:
|
|
res = db.create_record(self.base_id, self.table_id, fields)
|
|
record_id = res['id']
|
|
|
|
db.upload_attachment(self.base_id, record_id, 'GCODE', temp_path)
|
|
return str(record_id)
|