713 lines
25 KiB
Python
713 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import io
|
|
import json
|
|
import signal
|
|
import os
|
|
import re
|
|
import shlex
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from flask import Flask, jsonify, render_template, request, send_from_directory
|
|
from PIL import Image, ImageDraw, ImageFont, ImageOps
|
|
|
|
try:
|
|
import barcode
|
|
from barcode.writer import ImageWriter
|
|
except Exception: # pragma: no cover - optional at runtime until installed
|
|
barcode = None
|
|
ImageWriter = None
|
|
|
|
try:
|
|
import qrcode
|
|
except Exception: # pragma: no cover - optional at runtime until installed
|
|
qrcode = None
|
|
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent.parent
|
|
LABEL_DIR = Path(os.environ.get("LABEL_DIR", BASE_DIR / "labels"))
|
|
LABEL_DIR.mkdir(parents=True, exist_ok=True)
|
|
QUEUE_DIR = LABEL_DIR / "queue"
|
|
QUEUE_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
DEFAULT_PRINT_COMMAND = "ptouch-print --image {image}"
|
|
PRINT_COMMAND = os.environ.get("PRINT_COMMAND", DEFAULT_PRINT_COMMAND)
|
|
PRINT_ENABLED = os.environ.get("PRINT_ENABLED", "1").lower() in {"1", "true", "yes", "on"}
|
|
PRINT_QUEUE_ENABLED = os.environ.get("PRINT_QUEUE_ENABLED", "1").lower() in {"1", "true", "yes", "on"}
|
|
PRINT_QUEUE_RETRY_SECONDS = int(os.environ.get("PRINT_QUEUE_RETRY_SECONDS", "30"))
|
|
DPI = int(os.environ.get("LABEL_DPI", "180"))
|
|
PTOUCH_SUPPORTS_COPIES: bool | None = None
|
|
|
|
TAPE_WIDTHS_MM = [3.5, 6, 9, 12, 18, 24]
|
|
PRINTABLE_TAPE_PX = {
|
|
3.5: 24,
|
|
6.0: 32,
|
|
9.0: 52,
|
|
12.0: 76,
|
|
18.0: 120,
|
|
24.0: 128,
|
|
}
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = os.environ.get("SECRET_KEY", "change-me")
|
|
PRINT_LOCK = threading.Lock()
|
|
QUEUE_WAKE = threading.Event()
|
|
|
|
|
|
def mm_to_px(mm: float) -> int:
|
|
return max(1, round(mm / 25.4 * DPI))
|
|
|
|
|
|
def printable_px_for_tape(tape_width_mm: float) -> int:
|
|
normalized = float(tape_width_mm)
|
|
return PRINTABLE_TAPE_PX.get(normalized, mm_to_px(normalized))
|
|
|
|
|
|
def render_scale_for_tape(tape_width_mm: float) -> float:
|
|
return printable_px_for_tape(tape_width_mm) / float(tape_width_mm)
|
|
|
|
|
|
def scaled_mm_to_px(mm: float, scale: float) -> int:
|
|
return max(1, round(mm * scale))
|
|
|
|
|
|
def clamp(value: float, low: float, high: float) -> float:
|
|
return max(low, min(high, value))
|
|
|
|
|
|
def get_font(size_px: int, bold: bool = False) -> ImageFont.FreeTypeFont | ImageFont.ImageFont:
|
|
candidates = [
|
|
"/usr/share/fonts/dejavu/DejaVuSans-Bold.ttf" if bold else "/usr/share/fonts/dejavu/DejaVuSans.ttf",
|
|
"/usr/share/fonts/TTF/DejaVuSans-Bold.ttf" if bold else "/usr/share/fonts/TTF/DejaVuSans.ttf",
|
|
"/System/Library/Fonts/Supplemental/Arial Bold.ttf" if bold else "/System/Library/Fonts/Supplemental/Arial.ttf",
|
|
]
|
|
for path in candidates:
|
|
if Path(path).exists():
|
|
return ImageFont.truetype(path, size_px)
|
|
return ImageFont.load_default()
|
|
|
|
|
|
def new_label_image(width_px: int, height_px: int) -> Image.Image:
|
|
image = Image.new("P", (width_px, height_px), 0)
|
|
image.putpalette([255, 255, 255, 0, 0, 0] + [0, 0, 0] * 254)
|
|
return image
|
|
|
|
|
|
def object_box(obj: dict[str, Any], scale: float) -> tuple[int, int, int, int]:
|
|
x = scaled_mm_to_px(float(obj.get("x", 0)), scale)
|
|
y = scaled_mm_to_px(float(obj.get("y", 0)), scale)
|
|
w = scaled_mm_to_px(float(obj.get("w", 10)), scale)
|
|
h = scaled_mm_to_px(float(obj.get("h", 5)), scale)
|
|
return x, y, max(1, w), max(1, h)
|
|
|
|
|
|
def text_lines(text: str) -> list[str]:
|
|
return text.split("\n")
|
|
|
|
|
|
def draw_text(image: Image.Image, draw: ImageDraw.ImageDraw, obj: dict[str, Any], scale: float) -> None:
|
|
x, y, w, h = object_box(obj, scale)
|
|
text = str(obj.get("text", "Text"))
|
|
font_size = scaled_mm_to_px(float(obj.get("fontSizeMm", 5)), scale)
|
|
font = get_font(font_size, bool(obj.get("bold", False)))
|
|
align = obj.get("align", "left")
|
|
valign = obj.get("valign", "middle")
|
|
lines = text_lines(text)
|
|
boxes = [draw.textbbox((0, 0), line, font=font) for line in lines]
|
|
line_heights = [box[3] - box[1] for box in boxes]
|
|
gap = max(1, round(font_size * 0.15))
|
|
total_h = sum(line_heights) + max(0, len(lines) - 1) * gap
|
|
if valign == "top":
|
|
cy = y
|
|
elif valign == "bottom":
|
|
cy = y + h - total_h
|
|
else:
|
|
cy = y + (h - total_h) // 2
|
|
for line, box, line_h in zip(lines, boxes, line_heights):
|
|
text_w = box[2] - box[0]
|
|
if align == "center":
|
|
cx = x + (w - text_w) // 2
|
|
elif align == "right":
|
|
cx = x + w - text_w
|
|
else:
|
|
cx = x
|
|
text_layer = Image.new("1", image.size, 0)
|
|
text_draw = ImageDraw.Draw(text_layer)
|
|
text_draw.text((cx, cy - box[1]), line, font=font, fill=1)
|
|
clip = Image.new("1", image.size, 0)
|
|
clip_draw = ImageDraw.Draw(clip)
|
|
clip_draw.rectangle([x, y, x + w, y + h], fill=1)
|
|
clipped = Image.new("1", image.size, 0)
|
|
clipped.paste(text_layer, mask=clip)
|
|
image.paste(1, mask=clipped)
|
|
cy += line_h + gap
|
|
|
|
|
|
def draw_shape(draw: ImageDraw.ImageDraw, obj: dict[str, Any], scale: float) -> None:
|
|
x, y, w, h = object_box(obj, scale)
|
|
line = max(1, scaled_mm_to_px(float(obj.get("lineMm", 0.25)), scale))
|
|
shape = obj.get("shape", "rect")
|
|
fill = 1 if obj.get("fill", False) else None
|
|
box = [x, y, x + w, y + h]
|
|
if shape == "ellipse":
|
|
draw.ellipse(box, outline=1, fill=fill, width=line)
|
|
elif shape == "line":
|
|
draw.line((x, y, x + w, y + h), fill=1, width=line)
|
|
else:
|
|
draw.rectangle(box, outline=1, fill=fill, width=line)
|
|
|
|
|
|
def paste_mono(image: Image.Image, source: Image.Image, x: int, y: int, w: int, h: int) -> None:
|
|
source = ImageOps.grayscale(source).resize((w, h), Image.Resampling.LANCZOS)
|
|
mask = source.point(lambda p: 255 if p < 160 else 0, "1")
|
|
black = Image.new("P", (w, h), 1)
|
|
black.putpalette([255, 255, 255, 0, 0, 0] + [0, 0, 0] * 254)
|
|
image.paste(black, (x, y), mask)
|
|
|
|
|
|
def decode_data_image(data_url: str) -> Image.Image:
|
|
if "," in data_url:
|
|
data_url = data_url.split(",", 1)[1]
|
|
return Image.open(io.BytesIO(base64.b64decode(data_url)))
|
|
|
|
|
|
def draw_qr(image: Image.Image, obj: dict[str, Any], scale: float) -> None:
|
|
if qrcode is None:
|
|
raise RuntimeError("qrcode ist nicht installiert.")
|
|
x, y, w, h = object_box(obj, scale)
|
|
qr = qrcode.QRCode(border=1, box_size=10)
|
|
qr.add_data(str(obj.get("text", "https://example.local")))
|
|
qr.make(fit=True)
|
|
qr_img = qr.make_image(fill_color="black", back_color="white").convert("RGB")
|
|
side = min(w, h)
|
|
paste_mono(image, qr_img, x, y, side, side)
|
|
|
|
|
|
def draw_barcode(image: Image.Image, obj: dict[str, Any], scale: float) -> None:
|
|
if barcode is None or ImageWriter is None:
|
|
raise RuntimeError("python-barcode ist nicht installiert.")
|
|
x, y, w, h = object_box(obj, scale)
|
|
writer = ImageWriter()
|
|
code = barcode.get("code128", str(obj.get("text", "1234567890")), writer=writer)
|
|
buffer = io.BytesIO()
|
|
code.write(buffer, {"write_text": bool(obj.get("showText", True)), "quiet_zone": 1.0, "module_height": 8.0})
|
|
buffer.seek(0)
|
|
paste_mono(image, Image.open(buffer), x, y, w, h)
|
|
|
|
|
|
def draw_table(draw: ImageDraw.ImageDraw, obj: dict[str, Any], scale: float) -> None:
|
|
x, y, w, h = object_box(obj, scale)
|
|
rows = max(1, int(obj.get("rows", 2)))
|
|
cols = max(1, int(obj.get("cols", 2)))
|
|
line = max(1, scaled_mm_to_px(float(obj.get("lineMm", 0.2)), scale))
|
|
draw.rectangle([x, y, x + w, y + h], outline=1, width=line)
|
|
for col in range(1, cols):
|
|
cx = x + round(w * col / cols)
|
|
draw.line((cx, y, cx, y + h), fill=1, width=line)
|
|
for row in range(1, rows):
|
|
cy = y + round(h * row / rows)
|
|
draw.line((x, cy, x + w, cy), fill=1, width=line)
|
|
|
|
|
|
def draw_cutmark(draw: ImageDraw.ImageDraw, width_px: int, height_px: int, scale: float) -> None:
|
|
x = max(0, width_px - scaled_mm_to_px(1.5, scale))
|
|
dash = max(2, scaled_mm_to_px(1, scale))
|
|
gap = max(2, scaled_mm_to_px(0.7, scale))
|
|
y = 0
|
|
while y < height_px:
|
|
draw.line((x, y, x, min(height_px, y + dash)), fill=1, width=1)
|
|
y += dash + gap
|
|
|
|
|
|
def crop_label_to_content(image: Image.Image, design: dict[str, Any], scale: float) -> Image.Image:
|
|
if not bool(design.get("autoLength", True)):
|
|
return image
|
|
|
|
tape_width = float(design.get("tapeWidthMm", 24))
|
|
orientation = design.get("orientation", "landscape")
|
|
margin_px = scaled_mm_to_px(float(design.get("marginMm", 2)), scale)
|
|
min_length_px = scaled_mm_to_px(10, scale)
|
|
black_mask = image.convert("L").point(lambda pixel: 255 if pixel < 128 else 0, "L")
|
|
bbox = black_mask.getbbox()
|
|
|
|
if orientation == "portrait":
|
|
fixed_width = scaled_mm_to_px(tape_width, scale)
|
|
if not bbox:
|
|
return new_label_image(fixed_width, min_length_px)
|
|
end = min(image.height, bbox[3] + margin_px)
|
|
target = min(image.height, max(min_length_px, end))
|
|
return image.crop((0, 0, image.width, target))
|
|
|
|
fixed_height = scaled_mm_to_px(tape_width, scale)
|
|
if not bbox:
|
|
return new_label_image(min_length_px, fixed_height)
|
|
end = min(image.width, bbox[2] + margin_px)
|
|
target = min(image.width, max(min_length_px, end))
|
|
return image.crop((0, 0, target, image.height))
|
|
|
|
|
|
def render_design_image(design: dict[str, Any]) -> Image.Image:
|
|
tape_width = float(design.get("tapeWidthMm", 24))
|
|
if tape_width not in TAPE_WIDTHS_MM:
|
|
raise ValueError("Ungueltige Bandbreite.")
|
|
length = clamp(float(design.get("lengthMm", 70)), 10, 500)
|
|
orientation = design.get("orientation", "landscape")
|
|
if orientation == "portrait":
|
|
canvas_w_mm, canvas_h_mm = tape_width, length
|
|
else:
|
|
canvas_w_mm, canvas_h_mm = length, tape_width
|
|
|
|
scale = render_scale_for_tape(tape_width)
|
|
image = new_label_image(scaled_mm_to_px(canvas_w_mm, scale), scaled_mm_to_px(canvas_h_mm, scale))
|
|
max_height_px = printable_px_for_tape(tape_width)
|
|
if image.height > max_height_px:
|
|
raise ValueError(
|
|
f"Gerendertes Etikett ist zu hoch ({image.height}px). "
|
|
f"Maximal fuer {tape_width:g} mm Band sind {max_height_px}px."
|
|
)
|
|
draw = ImageDraw.Draw(image)
|
|
|
|
for obj in design.get("objects", []):
|
|
if not isinstance(obj, dict) or obj.get("hidden"):
|
|
continue
|
|
obj_type = obj.get("type")
|
|
if obj_type == "text" or obj_type == "symbol":
|
|
draw_text(image, draw, obj, scale)
|
|
elif obj_type == "shape":
|
|
draw_shape(draw, obj, scale)
|
|
elif obj_type == "line":
|
|
draw_shape(draw, {**obj, "shape": "line"}, scale)
|
|
elif obj_type == "table":
|
|
draw_table(draw, obj, scale)
|
|
elif obj_type == "image" and obj.get("dataUrl"):
|
|
x, y, w, h = object_box(obj, scale)
|
|
paste_mono(image, decode_data_image(str(obj["dataUrl"])), x, y, w, h)
|
|
elif obj_type == "qr":
|
|
draw_qr(image, obj, scale)
|
|
elif obj_type == "barcode":
|
|
draw_barcode(image, obj, scale)
|
|
|
|
image = crop_label_to_content(image, design, scale)
|
|
draw = ImageDraw.Draw(image)
|
|
if design.get("cutMode") in {"cutmark", "cutmark-chain"}:
|
|
draw_cutmark(draw, image.width, image.height, scale)
|
|
|
|
return image
|
|
|
|
|
|
def render_design(design: dict[str, Any], copies: int = 1) -> Path:
|
|
image = render_design_image(design)
|
|
if copies > 1:
|
|
combined = new_label_image(image.width * copies, image.height)
|
|
for index in range(copies):
|
|
combined.paste(image, (index * image.width, 0))
|
|
image = combined
|
|
|
|
output = LABEL_DIR / f"label-{uuid.uuid4().hex}.png"
|
|
image.save(output)
|
|
return output
|
|
|
|
|
|
def ptouch_supports_copies() -> bool:
|
|
global PTOUCH_SUPPORTS_COPIES
|
|
if PTOUCH_SUPPORTS_COPIES is not None:
|
|
return PTOUCH_SUPPORTS_COPIES
|
|
result = subprocess.run(
|
|
["ptouch-print", "--help"],
|
|
capture_output=True,
|
|
check=False,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
help_text = f"{result.stdout}\n{result.stderr}"
|
|
PTOUCH_SUPPORTS_COPIES = "--copies" in help_text
|
|
return PTOUCH_SUPPORTS_COPIES
|
|
|
|
|
|
def cut_args_for_design(design: dict[str, Any], copies: int) -> str:
|
|
args = []
|
|
cut_mode = str(design.get("cutMode", "auto"))
|
|
if cut_mode in {"chain", "cutmark-chain"} and copies <= 1:
|
|
args.append("--chain")
|
|
return " ".join(args)
|
|
|
|
|
|
def run_print_command(image_path: Path, copies: int, design: dict[str, Any]) -> subprocess.CompletedProcess[str]:
|
|
cut_args = cut_args_for_design(design, copies)
|
|
command_copies = copies if "{copies}" in PRINT_COMMAND else 1
|
|
command = PRINT_COMMAND.format(image=shlex.quote(str(image_path)), copies=command_copies, cut_args=cut_args)
|
|
if copies > 1 and "{copies}" not in PRINT_COMMAND and ptouch_supports_copies():
|
|
command = f"{command} --copies {copies}"
|
|
if cut_args and "{cut_args}" not in PRINT_COMMAND:
|
|
command = f"{command} {cut_args}"
|
|
process = subprocess.Popen(
|
|
shlex.split(command),
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
start_new_session=True,
|
|
)
|
|
try:
|
|
stdout, stderr = process.communicate(timeout=90)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
stdout, stderr = process.communicate()
|
|
return subprocess.CompletedProcess(
|
|
shlex.split(command),
|
|
124,
|
|
stdout,
|
|
f"{stderr}\nDruckprozess wurde nach Timeout beendet.".strip(),
|
|
)
|
|
return subprocess.CompletedProcess(shlex.split(command), process.returncode, stdout, stderr)
|
|
|
|
|
|
def print_label(image_path: Path, copies: int, design: dict[str, Any]) -> list[subprocess.CompletedProcess[str]]:
|
|
return [run_print_command(image_path, copies, design)]
|
|
|
|
|
|
def run_print_job(image_path: Path, copies: int, design: dict[str, Any]) -> None:
|
|
if not PRINT_LOCK.acquire(blocking=False):
|
|
raise RuntimeError("Es laeuft bereits ein Druckjob. Bitte warten oder haengenden ptouch-print-Prozess beenden.")
|
|
try:
|
|
results = print_label(image_path, copies, design)
|
|
failed = next((result for result in results if result.returncode != 0), None)
|
|
if failed:
|
|
raise RuntimeError(format_print_error(failed))
|
|
finally:
|
|
PRINT_LOCK.release()
|
|
|
|
|
|
def queued_job_path(job_id: str) -> Path:
|
|
return QUEUE_DIR / f"{job_id}.json"
|
|
|
|
|
|
def enqueue_print_job(image_path: Path, copies: int, design: dict[str, Any]) -> dict[str, Any]:
|
|
now = time.time()
|
|
job = {
|
|
"id": uuid.uuid4().hex,
|
|
"image": image_path.name,
|
|
"copies": copies,
|
|
"design": design,
|
|
"createdAt": now,
|
|
"nextAttemptAt": now,
|
|
"attempts": 0,
|
|
"lastError": "",
|
|
}
|
|
target = queued_job_path(job["id"])
|
|
temp = target.with_suffix(".tmp")
|
|
temp.write_text(json.dumps(job, ensure_ascii=True, indent=2), encoding="utf-8")
|
|
temp.replace(target)
|
|
QUEUE_WAKE.set()
|
|
return job
|
|
|
|
|
|
def load_queue_jobs() -> list[dict[str, Any]]:
|
|
jobs: list[dict[str, Any]] = []
|
|
for path in sorted(QUEUE_DIR.glob("*.json")):
|
|
try:
|
|
job = json.loads(path.read_text(encoding="utf-8"))
|
|
if isinstance(job, dict):
|
|
jobs.append(job)
|
|
except Exception:
|
|
continue
|
|
return sorted(jobs, key=lambda item: float(item.get("createdAt", 0)))
|
|
|
|
|
|
def save_queue_job(job: dict[str, Any]) -> None:
|
|
queued_job_path(str(job["id"])).write_text(json.dumps(job, ensure_ascii=True, indent=2), encoding="utf-8")
|
|
|
|
|
|
def process_queue_once() -> None:
|
|
if not PRINT_ENABLED:
|
|
return
|
|
now = time.time()
|
|
for job in load_queue_jobs():
|
|
if float(job.get("nextAttemptAt", 0)) > now:
|
|
continue
|
|
image_path = LABEL_DIR / str(job.get("image", ""))
|
|
try:
|
|
if not image_path.exists():
|
|
raise RuntimeError(f"Queue-Bild fehlt: {image_path.name}")
|
|
run_print_job(image_path, int(job.get("copies", 1)), job.get("design") if isinstance(job.get("design"), dict) else {})
|
|
queued_job_path(str(job["id"])).unlink(missing_ok=True)
|
|
except Exception as exc:
|
|
job["attempts"] = int(job.get("attempts", 0)) + 1
|
|
job["lastError"] = str(exc)
|
|
job["nextAttemptAt"] = time.time() + PRINT_QUEUE_RETRY_SECONDS
|
|
save_queue_job(job)
|
|
break
|
|
|
|
|
|
def print_queue_worker() -> None:
|
|
while True:
|
|
try:
|
|
process_queue_once()
|
|
except Exception:
|
|
pass
|
|
QUEUE_WAKE.wait(PRINT_QUEUE_RETRY_SECONDS)
|
|
QUEUE_WAKE.clear()
|
|
|
|
|
|
def should_render_copies_as_strip(design: dict[str, Any], copies: int) -> bool:
|
|
if copies <= 1:
|
|
return False
|
|
return str(design.get("cutMode", "auto")) in {"chain", "cutmark-chain"}
|
|
|
|
|
|
COLOR_HEX_BY_NAME = {
|
|
"white": "#ffffff",
|
|
"black": "#111111",
|
|
"red": "#d92d20",
|
|
"blue": "#2563eb",
|
|
"yellow": "#facc15",
|
|
"green": "#16a34a",
|
|
"clear": "#f8fafc",
|
|
"transparent": "#f8fafc",
|
|
"silver": "#c0c0c0",
|
|
"gold": "#d4af37",
|
|
"matte white": "#f5f5f0",
|
|
"matte black": "#1f1f1f",
|
|
}
|
|
|
|
|
|
COLOR_NAME_DE = {
|
|
"white": "Weiss",
|
|
"black": "Schwarz",
|
|
"red": "Rot",
|
|
"blue": "Blau",
|
|
"yellow": "Gelb",
|
|
"green": "Gruen",
|
|
"clear": "Transparent",
|
|
"transparent": "Transparent",
|
|
"silver": "Silber",
|
|
"gold": "Gold",
|
|
"matte white": "Matt weiss",
|
|
"matte black": "Matt schwarz",
|
|
}
|
|
|
|
|
|
MEDIA_TYPE_DE = {
|
|
"laminated tape": "Laminiertes Band",
|
|
"non-laminated tape": "Nicht laminiertes Band",
|
|
"fabric tape": "Textilband",
|
|
"heat shrink tube": "Schrumpfschlauch",
|
|
}
|
|
|
|
|
|
def color_info_from_line(output: str, key: str) -> dict[str, str] | None:
|
|
match = re.search(rf"^{re.escape(key)}\s*=\s*([0-9a-fA-F]+)(?:\s*\(([^)]+)\))?", output, re.MULTILINE)
|
|
if not match:
|
|
return None
|
|
code = match.group(1)
|
|
raw_name = (match.group(2) or f"Code {code}").strip()
|
|
normalized = raw_name.lower()
|
|
return {
|
|
"code": code,
|
|
"name": COLOR_NAME_DE.get(normalized, raw_name),
|
|
"rawName": raw_name,
|
|
"css": COLOR_HEX_BY_NAME.get(normalized, "#d1d5db"),
|
|
}
|
|
|
|
|
|
def media_type_from_output(output: str) -> dict[str, str] | None:
|
|
match = re.search(r"^media type\s*=\s*([0-9a-fA-F]+)(?:\s*\(([^)]+)\))?", output, re.MULTILINE)
|
|
if not match:
|
|
return None
|
|
code = match.group(1)
|
|
raw_name = (match.group(2) or f"Code {code}").strip()
|
|
normalized = raw_name.lower()
|
|
return {
|
|
"code": code,
|
|
"name": MEDIA_TYPE_DE.get(normalized, raw_name),
|
|
"rawName": raw_name,
|
|
}
|
|
|
|
|
|
def printable_width_from_output(output: str) -> int | None:
|
|
match = re.search(r"maximum printing width[^\n]*?(\d+)\s*px", output, re.IGNORECASE)
|
|
if not match:
|
|
return None
|
|
return int(match.group(1))
|
|
|
|
|
|
def detect_tape_info() -> dict[str, Any]:
|
|
result = subprocess.run(
|
|
["ptouch-print", "--info"],
|
|
capture_output=True,
|
|
check=False,
|
|
text=True,
|
|
timeout=20,
|
|
)
|
|
output = "\n".join(part for part in [result.stdout.strip(), result.stderr.strip()] if part)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(format_print_error(result))
|
|
|
|
width = None
|
|
candidates: list[float] = []
|
|
for pattern in [
|
|
r"(?:tape|media|band|width|breite|size|groesse|printing)[^\n:=-]*[:=-]?\s*(\d+(?:[.,]\d+)?)\s*mm",
|
|
r"(\d+(?:[.,]\d+)?)\s*mm",
|
|
]:
|
|
for match in re.finditer(pattern, output, re.IGNORECASE):
|
|
candidates.append(float(match.group(1).replace(",", ".")))
|
|
|
|
for supported in TAPE_WIDTHS_MM:
|
|
if any(abs(candidate - supported) < 0.26 for candidate in candidates):
|
|
width = supported
|
|
break
|
|
|
|
return {
|
|
"width": width,
|
|
"printableWidthPx": printable_width_from_output(output),
|
|
"mediaType": media_type_from_output(output),
|
|
"tapeColor": color_info_from_line(output, "tape color"),
|
|
"textColor": color_info_from_line(output, "text color"),
|
|
"raw": output,
|
|
}
|
|
|
|
|
|
def format_print_error(result: subprocess.CompletedProcess[str]) -> str:
|
|
details = result.stderr.strip() or result.stdout.strip() or "Unbekannter Fehler"
|
|
if "PLite Mode" in details or "RemovableDisk" in details:
|
|
return (
|
|
"Drucker ist im Editor-Lite/PLite-Modus. Am PT-P700 die Editor-Lite-Taste deaktivieren, "
|
|
"bis die Editor-Lite-LED aus ist, USB neu verbinden und erneut drucken. "
|
|
f"Details: {details}"
|
|
)
|
|
if "LIBUSB_ERROR_IO" in details or "libusb_open error" in details:
|
|
return (
|
|
"Drucker wurde gefunden, konnte aber per libusb nicht geoeffnet werden. "
|
|
"In Unraid pruefen: Privileged an, /dev/bus/usb Read/Write gemountet, usblp nicht geladen. "
|
|
f"Details: {details}"
|
|
)
|
|
if "timeout while waiting for status response" in details:
|
|
return (
|
|
"Drucker wurde gefunden, antwortet aber nicht auf die Statusabfrage. "
|
|
"PT-P700 kurz ausschalten oder USB abziehen, usblp auf dem Host erneut pruefen/entladen "
|
|
"und danach den Container neu starten. "
|
|
f"Details: {details}"
|
|
)
|
|
if result.returncode == 124 or "Druckprozess wurde nach Timeout beendet" in details:
|
|
return (
|
|
"Druckprozess hing und wurde beendet. "
|
|
"Bitte auf dem Host pruefen, ob kein ptouch-print mehr das USB-Geraet blockiert, "
|
|
"danach Drucker kurz neu verbinden und erneut versuchen. "
|
|
f"Details: {details}"
|
|
)
|
|
return f"Druck fehlgeschlagen: {details}"
|
|
|
|
|
|
def design_from_request() -> tuple[dict[str, Any], int]:
|
|
payload = request.get_json(force=True)
|
|
design = payload.get("design") if isinstance(payload, dict) else None
|
|
if not isinstance(design, dict):
|
|
raise ValueError("Kein gueltiges Layout erhalten.")
|
|
copies = int(payload.get("copies", 1))
|
|
return design, max(1, min(20, copies))
|
|
|
|
|
|
@app.get("/")
|
|
def index():
|
|
return render_template(
|
|
"index.html",
|
|
tape_widths=TAPE_WIDTHS_MM,
|
|
print_enabled=PRINT_ENABLED,
|
|
default_command=PRINT_COMMAND,
|
|
)
|
|
|
|
|
|
@app.get("/mobile")
|
|
def mobile():
|
|
return render_template(
|
|
"mobile.html",
|
|
tape_widths=TAPE_WIDTHS_MM,
|
|
print_enabled=PRINT_ENABLED,
|
|
default_command=PRINT_COMMAND,
|
|
)
|
|
|
|
|
|
@app.post("/api/preview")
|
|
def api_preview():
|
|
try:
|
|
design, _copies = design_from_request()
|
|
image_path = render_design(design, 1)
|
|
return jsonify({"ok": True, "image": f"/labels/{image_path.name}"})
|
|
except Exception as exc:
|
|
return jsonify({"ok": False, "error": str(exc)}), 400
|
|
|
|
|
|
@app.post("/api/print")
|
|
def api_print():
|
|
try:
|
|
design, copies = design_from_request()
|
|
render_copies = copies if should_render_copies_as_strip(design, copies) else 1
|
|
print_copies = 1 if render_copies > 1 else copies
|
|
image_path = render_design(design, render_copies)
|
|
print_design = design
|
|
if render_copies > 1 and str(design.get("cutMode", "auto")) in {"chain", "cutmark-chain"}:
|
|
print_design = dict(design)
|
|
print_design["cutMode"] = "cutmark" if design.get("cutMode") == "cutmark-chain" else "auto"
|
|
if PRINT_ENABLED:
|
|
if PRINT_QUEUE_ENABLED:
|
|
job = enqueue_print_job(image_path, print_copies, print_design)
|
|
queue_len = len(load_queue_jobs())
|
|
message = f"Etikett wurde in die Druckwarteschlange gelegt. Position: {queue_len}."
|
|
return jsonify({"ok": True, "message": message, "image": f"/labels/{image_path.name}", "queued": True, "jobId": job["id"], "queueLength": queue_len})
|
|
run_print_job(image_path, print_copies, print_design)
|
|
message = "Etikett wurde an den Drucker gesendet."
|
|
else:
|
|
message = f"Testmodus: Etikett erstellt, nicht gedruckt: {image_path.name}"
|
|
return jsonify({"ok": True, "message": message, "image": f"/labels/{image_path.name}"})
|
|
except Exception as exc:
|
|
return jsonify({"ok": False, "error": str(exc)}), 400
|
|
|
|
|
|
@app.get("/api/tape-info")
|
|
def api_tape_info():
|
|
try:
|
|
info = detect_tape_info()
|
|
if info["width"] is None:
|
|
return jsonify({"ok": False, "error": "Bandbreite konnte nicht erkannt werden.", **info}), 400
|
|
return jsonify({"ok": True, **info})
|
|
except Exception as exc:
|
|
return jsonify({"ok": False, "error": str(exc)}), 400
|
|
|
|
|
|
@app.get("/api/queue")
|
|
def api_queue():
|
|
jobs = load_queue_jobs()
|
|
return jsonify(
|
|
{
|
|
"ok": True,
|
|
"enabled": PRINT_QUEUE_ENABLED,
|
|
"length": len(jobs),
|
|
"jobs": [
|
|
{
|
|
"id": job.get("id"),
|
|
"image": job.get("image"),
|
|
"copies": job.get("copies"),
|
|
"attempts": job.get("attempts", 0),
|
|
"lastError": job.get("lastError", ""),
|
|
"createdAt": job.get("createdAt"),
|
|
"nextAttemptAt": job.get("nextAttemptAt"),
|
|
}
|
|
for job in jobs
|
|
],
|
|
}
|
|
)
|
|
|
|
|
|
@app.get("/labels/<path:filename>")
|
|
def labels(filename: str):
|
|
return send_from_directory(LABEL_DIR, filename)
|
|
|
|
|
|
if PRINT_QUEUE_ENABLED:
|
|
threading.Thread(target=print_queue_worker, name="ptouch-print-queue", daemon=True).start()
|