Jonas MenzelandClaude Opus 5.5
bf32667f77
- Reads /opt/usr/canvas_record_info.json over SSH (pinned host key, restricted key), decodes Spoolman spool ids from the tag weight - Mirrors spools into CC2 Control's spool library and assigns slots, keeps its own slot mapping, sets Spoolman locations - Books the drop of CC2 Control's measured remaining weight to Spoolman after each print, idempotent across restarts and interrupted bookings - Dry-run by default; web UI for settings, SSH key, status and log - Tag generator for NFC Tools command chains (NTAG213/215/216) - Tests, mock environment, Dockerfile, Gitea Actions CI Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014GcEk5tkfFSYD6NTF66NkC
268 lines
9.4 KiB
Python
268 lines
9.4 KiB
Python
"""Encode/decode Elegoo Canvas NFC tags (NTAG213/215/216) and canvas records.
|
|
|
|
Page numbers are hex, as used by NFC Tools ("A2:PP:XXXXXXXX"). All values are
|
|
big-endian. Pages 0-3 (UID, lock bits, capability container) are never written.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import colorsys
|
|
import math
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
|
|
from .materials import MATERIAL_BY_CODE, MATERIALS, SUBTYPE_BY_CODE, material_of_subtype
|
|
|
|
DEFAULT_URL = "https://www.elegoo.com"
|
|
DEFAULT_ID_OFFSET = 10000
|
|
MAX_WEIGHT = 0xFFFF
|
|
|
|
NDEF_FIRST_PAGE = 0x04
|
|
NDEF_LAST_PAGE = 0x0F # canvas data starts at page 0x10
|
|
CANVAS_PAGES = range(0x10, 0x19)
|
|
|
|
PAGE_18_DEFAULT = 0x0036C800 # copied from original Elegoo spools; meaning unknown
|
|
|
|
# Display palette from /opt/inst/filament_color_set.json
|
|
PALETTE = [
|
|
"FFFFFF", "FFF242", "DBF47A", "09CC3A", "077747", "0B6283", "0BE2A0", "74D9F3",
|
|
"48A7FA", "2850DF", "433089", "A03BF7", "F32FF8", "D4B1DD", "F95D77", "F72221",
|
|
"7C4C00", "F88D36", "FCEBD7", "D2C5A3", "AF7832", "898989", "BCBCBC", "000000",
|
|
]
|
|
BLUE = "2850DF"
|
|
|
|
_URI_PREFIXES = [
|
|
("https://www.", 0x02), ("http://www.", 0x01), ("https://", 0x04), ("http://", 0x03),
|
|
]
|
|
|
|
|
|
class TagError(ValueError):
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class TagData:
|
|
spool_id: int
|
|
color: str # RRGGBB
|
|
material: str = "PLA"
|
|
subtype: int = 0x0001
|
|
temp_min: int = 190
|
|
temp_max: int = 230
|
|
diameter: float = 1.75
|
|
finish: int = 0xFF
|
|
filament_code: int = 0x0000
|
|
id_offset: int = DEFAULT_ID_OFFSET
|
|
url: str = DEFAULT_URL
|
|
warnings: list[str] = field(default_factory=list)
|
|
|
|
|
|
def normalize_hex_color(value: str | None) -> str | None:
|
|
if not value:
|
|
return None
|
|
v = value.strip().lstrip("#").upper()
|
|
if re.fullmatch(r"[0-9A-F]{8}", v): # RRGGBBAA
|
|
v = v[:6]
|
|
if re.fullmatch(r"[0-9A-F]{3}", v):
|
|
v = "".join(c * 2 for c in v)
|
|
return v if re.fullmatch(r"[0-9A-F]{6}", v) else None
|
|
|
|
|
|
# ---------- NDEF ----------
|
|
|
|
def encode_ndef_url(url: str) -> bytes:
|
|
"""Lock-control TLV + NDEF URI record + terminator, as on the original tags."""
|
|
prefix_code, rest = 0x00, url
|
|
for prefix, code in _URI_PREFIXES:
|
|
if url.startswith(prefix):
|
|
prefix_code, rest = code, url[len(prefix):]
|
|
break
|
|
body = rest.encode("utf-8")
|
|
payload = bytes([prefix_code]) + body
|
|
if len(payload) > 255:
|
|
raise TagError("URL zu lang")
|
|
record = bytes([0xD1, 0x01, len(payload), 0x55]) + payload
|
|
data = bytes([0x01, 0x03, 0xA0, 0x0C, 0x34, 0x03, len(record)]) + record + b"\xFE"
|
|
max_len = (NDEF_LAST_PAGE - NDEF_FIRST_PAGE + 1) * 4
|
|
if len(data) > max_len:
|
|
raise TagError(f"URL zu lang: {len(data)} Byte, maximal {max_len} (Seiten 04-0F)")
|
|
pad = (-len(data)) % 4
|
|
return data + b"\x00" * pad
|
|
|
|
|
|
def decode_ndef_url(data: bytes) -> str | None:
|
|
i = 0
|
|
while i < len(data):
|
|
t = data[i]
|
|
if t == 0x00:
|
|
i += 1
|
|
continue
|
|
if t == 0xFE or i + 1 >= len(data):
|
|
return None
|
|
length = data[i + 1]
|
|
value = data[i + 2:i + 2 + length]
|
|
if t == 0x03 and len(value) >= 5 and value[3] == 0x55:
|
|
plen = value[2]
|
|
payload = value[4:4 + plen]
|
|
prefix = {code: p for p, code in _URI_PREFIXES}.get(payload[0], "")
|
|
return prefix + payload[1:].decode("utf-8", "replace")
|
|
i += 2 + length
|
|
return None
|
|
|
|
|
|
# ---------- canvas pages ----------
|
|
|
|
def encode_pages(tag: TagData) -> dict[int, bytes]:
|
|
if tag.material not in MATERIALS:
|
|
raise TagError(f"Unbekanntes Material {tag.material}")
|
|
if tag.subtype not in SUBTYPE_BY_CODE:
|
|
raise TagError(f"Unbekannter Untertyp 0x{tag.subtype:04X}")
|
|
if material_of_subtype(tag.subtype) != tag.material:
|
|
raise TagError(f"Untertyp {SUBTYPE_BY_CODE[tag.subtype]} passt nicht zu {tag.material}")
|
|
color = normalize_hex_color(tag.color)
|
|
if color is None:
|
|
raise TagError("Farbe muss RRGGBB sein")
|
|
weight = tag.spool_id + tag.id_offset
|
|
if tag.spool_id < 0 or weight > MAX_WEIGHT:
|
|
raise TagError(f"Spulen-ID {tag.spool_id} passt nicht in 2 Byte (max. {MAX_WEIGHT - tag.id_offset})")
|
|
for t in (tag.temp_min, tag.temp_max):
|
|
if not 0 <= t <= 500:
|
|
raise TagError("Düsentemperatur außerhalb 0-500 °C")
|
|
if tag.temp_min > tag.temp_max:
|
|
raise TagError("Düse min ist größer als max")
|
|
diameter = round(tag.diameter * 100)
|
|
if not 0 < diameter <= 0xFFFF:
|
|
raise TagError("Ungültiger Durchmesser")
|
|
|
|
pages: dict[int, bytes] = {}
|
|
ndef = encode_ndef_url(tag.url)
|
|
for i in range(0, len(ndef), 4):
|
|
pages[NDEF_FIRST_PAGE + i // 4] = ndef[i:i + 4]
|
|
pages[0x10] = bytes([0x36, 0xEE, 0xEE, 0xEE])
|
|
pages[0x11] = bytes([0xEE]) + tag.filament_code.to_bytes(2, "big") + b"\x00"
|
|
pages[0x12] = MATERIALS[tag.material]["code"].to_bytes(4, "big")
|
|
pages[0x13] = tag.subtype.to_bytes(2, "big") + b"\x00\x00"
|
|
pages[0x14] = bytes.fromhex(color) + bytes([tag.finish & 0xFF])
|
|
pages[0x15] = tag.temp_min.to_bytes(2, "big") + tag.temp_max.to_bytes(2, "big")
|
|
pages[0x16] = b"\x00\x00\x00\x00"
|
|
pages[0x17] = diameter.to_bytes(2, "big") + weight.to_bytes(2, "big")
|
|
pages[0x18] = PAGE_18_DEFAULT.to_bytes(4, "big")
|
|
return pages
|
|
|
|
|
|
def pages_to_commands(pages: dict[int, bytes]) -> str:
|
|
for p in pages:
|
|
if p < 4:
|
|
raise TagError("Seiten 0-3 dürfen nicht geschrieben werden")
|
|
return ",".join(f"A2:{p:02X}:{pages[p].hex().upper()}" for p in sorted(pages))
|
|
|
|
|
|
def encode_commands(tag: TagData) -> str:
|
|
return pages_to_commands(encode_pages(tag))
|
|
|
|
|
|
def parse_commands(text: str) -> dict[int, bytes]:
|
|
pages: dict[int, bytes] = {}
|
|
for part in re.split(r"[,\s]+", text.strip()):
|
|
if not part:
|
|
continue
|
|
m = re.fullmatch(r"A2:([0-9A-Fa-f]{2}):([0-9A-Fa-f]{8})", part)
|
|
if not m:
|
|
raise TagError(f"Ungültiger Befehl: {part}")
|
|
pages[int(m.group(1), 16)] = bytes.fromhex(m.group(2))
|
|
return pages
|
|
|
|
|
|
def decode_pages(pages: dict[int, bytes], id_offset: int = DEFAULT_ID_OFFSET) -> dict:
|
|
def page(n: int) -> bytes:
|
|
return pages.get(n, b"\x00\x00\x00\x00")
|
|
|
|
ndef = b"".join(page(p) for p in range(NDEF_FIRST_PAGE, NDEF_LAST_PAGE + 1))
|
|
mat_code = int.from_bytes(page(0x12), "big")
|
|
subtype = int.from_bytes(page(0x13)[:2], "big")
|
|
weight = int.from_bytes(page(0x17)[2:], "big")
|
|
return {
|
|
"url": decode_ndef_url(ndef),
|
|
"header": page(0x10)[0],
|
|
"manufacturer": page(0x10)[1:].hex().upper() + page(0x11)[:1].hex().upper(),
|
|
"filament_code": int.from_bytes(page(0x11)[1:3], "big"),
|
|
"material": MATERIAL_BY_CODE.get(mat_code, f"0x{mat_code:08X}"),
|
|
"subtype": subtype,
|
|
"subtype_name": SUBTYPE_BY_CODE.get(subtype, "?"),
|
|
"color": page(0x14)[:3].hex().upper(),
|
|
"finish": page(0x14)[3],
|
|
"temp_min": int.from_bytes(page(0x15)[:2], "big"),
|
|
"temp_max": int.from_bytes(page(0x15)[2:], "big"),
|
|
"diameter": int.from_bytes(page(0x17)[:2], "big") / 100,
|
|
"weight": weight,
|
|
"spool_id": spool_id_from_weight(weight, id_offset),
|
|
}
|
|
|
|
|
|
# ---------- canvas record (/opt/usr/canvas_record_info.json) ----------
|
|
|
|
def spool_id_from_weight(weight, id_offset: int = DEFAULT_ID_OFFSET) -> int | None:
|
|
"""weight >= offset encodes a Spoolman spool id; smaller values are real weights."""
|
|
try:
|
|
w = int(float(str(weight).strip()))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if w < id_offset or w > MAX_WEIGHT:
|
|
return None
|
|
return w - id_offset
|
|
|
|
|
|
def parse_record(record: dict, id_offset: int = DEFAULT_ID_OFFSET) -> list[dict]:
|
|
"""Return 4 slot dicts (slot 0-3) with the raw channel and the decoded spool id."""
|
|
slots = []
|
|
for n in range(4):
|
|
ch = record.get(f"channels{n}") or {}
|
|
if not isinstance(ch, dict):
|
|
ch = {}
|
|
occupied = any(str(v).strip() for v in ch.values())
|
|
slots.append({
|
|
"slot": n,
|
|
"channel": ch,
|
|
"occupied": occupied,
|
|
"spool_id": spool_id_from_weight(ch.get("weight"), id_offset) if occupied else None,
|
|
})
|
|
return slots
|
|
|
|
|
|
# ---------- colour suggestion ----------
|
|
|
|
def _srgb_to_lab(hexcolor: str) -> tuple[float, float, float]:
|
|
rgb = [int(hexcolor[i:i + 2], 16) / 255 for i in (0, 2, 4)]
|
|
lin = [c / 12.92 if c <= 0.04045 else ((c + 0.055) / 1.055) ** 2.4 for c in rgb]
|
|
r, g, b = lin
|
|
x = (0.4124 * r + 0.3576 * g + 0.1805 * b) / 0.95047
|
|
y = 0.2126 * r + 0.7152 * g + 0.0722 * b
|
|
z = (0.0193 * r + 0.1192 * g + 0.9505 * b) / 1.08883
|
|
|
|
def f(t):
|
|
return t ** (1 / 3) if t > 216 / 24389 else (24389 / 27 * t + 16) / 116
|
|
|
|
fx, fy, fz = f(x), f(y), f(z)
|
|
return 116 * fy - 16, 500 * (fx - fy), 200 * (fy - fz)
|
|
|
|
|
|
def _lab_distance(a: str, b: str) -> float:
|
|
return math.dist(_srgb_to_lab(a), _srgb_to_lab(b))
|
|
|
|
|
|
def _is_saturated_blue(hexcolor: str) -> bool:
|
|
r, g, b = (int(hexcolor[i:i + 2], 16) / 255 for i in (0, 2, 4))
|
|
h, s, v = colorsys.rgb_to_hsv(r, g, b)
|
|
return 215 <= h * 360 <= 245 and s >= 0.6 and v >= 0.35
|
|
|
|
|
|
def nearest_palette_color(hexcolor: str | None) -> str:
|
|
color = normalize_hex_color(hexcolor)
|
|
if color is None:
|
|
return "000000"
|
|
if color in PALETTE:
|
|
return color
|
|
if _is_saturated_blue(color):
|
|
# Lab distance lands on the violet entries for saturated blues.
|
|
return BLUE
|
|
return min(PALETTE, key=lambda p: _lab_distance(color, p))
|