Dominic Gundel
db6a85a703
Senden und Abbruch laufen über den Treiber, Orca-Doppelzeilen und tote Ideen-API fallen weg, Drucker folgen der Datei, Etikettenformat liegt in den Einstellungen.
126 lines
4.5 KiB
Python
126 lines
4.5 KiB
Python
"""Marlin-Dienst mit einer Leitung aus dem Test. Kein USB-Gerät."""
|
|
|
|
import sys
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from grotti.drivers.marlin import MarlinDriver
|
|
from grotti.marlin_daemon import serve
|
|
from grotti.marlin_link import MarlinService, gcode_lines, sd_filename, split_port
|
|
from grotti.moonraker import PrinterError
|
|
|
|
|
|
class ScriptedPort:
|
|
def __init__(self):
|
|
self.written = []
|
|
self._queue = []
|
|
|
|
def write(self, data):
|
|
text = data.decode().strip()
|
|
self.written.append(text)
|
|
if text.startswith("M27"):
|
|
self._queue.append(b"Done printing file\n")
|
|
self._queue.append(b"ok\n")
|
|
elif text.startswith("M105"):
|
|
self._queue.append(
|
|
b"ok T:200.0 /210.0 T0:200.0 /210.0 T1:190.0 /180.0 B:60.0 /60.0\n"
|
|
)
|
|
else:
|
|
self._queue.append(b"ok\n")
|
|
|
|
def readline(self):
|
|
if self._queue:
|
|
return self._queue.pop(0)
|
|
return b""
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
GCODE = b"; Kommentar\nT0\nM109 S210\nT1\nM109 S180\nG1 X1\n"
|
|
|
|
|
|
class MarlinTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.port = ScriptedPort()
|
|
self.service = MarlinService(open_port=lambda device, baud: self.port)
|
|
self.httpd = serve(0, self.service)
|
|
self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
base = "http://127.0.0.1:%s" % self.httpd.server_address[1]
|
|
self.drv = MarlinDriver(base=base)
|
|
self.printer = {
|
|
"name": "Tenlog",
|
|
"host": "COM5@250000",
|
|
"family": "tenlog",
|
|
"family_label": "Tenlog",
|
|
}
|
|
|
|
def tearDown(self):
|
|
self.httpd.shutdown()
|
|
self.httpd.server_close()
|
|
|
|
def test_split_and_short_name(self):
|
|
self.assertEqual(split_port("COM5@250000"), ("COM5", 250000))
|
|
self.assertEqual(sd_filename("teil drucken.gcode"), "TEILDRUC.GCO")
|
|
self.assertEqual(gcode_lines(GCODE), ["T0", "M109 S210", "T1", "M109 S180", "G1 X1"])
|
|
|
|
def test_two_heads_and_stream(self):
|
|
card = self.drv.snapshot([self.printer])[0]
|
|
self.assertTrue(card["ok"])
|
|
self.assertEqual(card["state"], "standby")
|
|
self.assertEqual(card["slots"], [])
|
|
self.assertEqual(len(card["tools"]), 2)
|
|
self.assertEqual(card["tools"][1]["temp"], 190.0)
|
|
self.assertEqual(card["host"], "COM5@250000")
|
|
stored = self.drv.upload("COM5@250000", "teil.gcode", GCODE)
|
|
self.drv.start_print("COM5@250000", stored, mode="stream")
|
|
self.service.session("COM5")._thread.join(3)
|
|
sent = self.port.written
|
|
self.assertIn("T0", sent)
|
|
self.assertIn("T1", sent)
|
|
self.assertNotIn("M28 teil.gcode", sent)
|
|
status = self.service.session("COM5").status()
|
|
self.assertEqual(status["state"], "complete")
|
|
self.assertEqual(status["progress"], 1)
|
|
|
|
def test_sd_writes_then_starts(self):
|
|
self.drv.attach(self.printer)
|
|
stored = self.drv.upload("COM5", "teil.gcode", GCODE)
|
|
shown = self.drv.start_print("COM5@250000", stored, mode="sd")
|
|
self.assertEqual(shown, "TEIL.GCO")
|
|
self.service.session("COM5")._thread.join(5)
|
|
sent = self.port.written
|
|
self.assertIn("M28 TEIL.GCO", sent)
|
|
self.assertIn("M29", sent)
|
|
self.assertIn("M23 TEIL.GCO", sent)
|
|
self.assertIn("M24", sent)
|
|
self.assertLess(sent.index("M28 TEIL.GCO"), sent.index("T0"))
|
|
self.assertLess(sent.index("M29"), sent.index("M24"))
|
|
|
|
def test_pause_and_cancel_two_heaters(self):
|
|
self.drv.attach(self.printer)
|
|
session = self.service.session("COM5")
|
|
session.state = "printing"
|
|
session.mode = "sd"
|
|
self.drv.command("COM5", "pause")
|
|
self.assertIn("M25", self.port.written)
|
|
self.assertEqual(session.state, "paused")
|
|
session.cancel()
|
|
self.assertIn("M104 T0 S0", self.port.written)
|
|
self.assertIn("M104 T1 S0", self.port.written)
|
|
self.assertIn("M140 S0", self.port.written)
|
|
self.assertEqual(session.state, "standby")
|
|
|
|
def test_without_service_stays_offline(self):
|
|
drv = MarlinDriver(base="http://127.0.0.1:9")
|
|
card = drv.snapshot([{"name": "i3", "host": "COM9", "family": "anycubic_i3"}])[0]
|
|
self.assertFalse(card["ok"])
|
|
self.assertEqual(card["state"], "offline")
|
|
with self.assertRaises(PrinterError):
|
|
drv.start_print("COM9", "a.gcode")
|